leadsdb 0.9.0

Python SDK for the LeadsDB API

leadsdb

Python SDK for the LeadsDB API.

Features:

• Sync and async clients
• Pydantic models with validation
• Fluent filter API
• Automatic pagination
• Retry with exponential backoff
• Full type hints

Installation

pip install leadsdb

Requires Python 3.10+.

Quick Start

from leadsdb import LeadsDB, Lead

client = LeadsDB("your-api-key")

# Create a lead
lead = client.create(Lead(
    name="Acme Corporation",
    source="website",
    city="San Francisco",
    email="[e-mail]",
    rating=4.5,
    tags=["enterprise", "saas"],
))

print(f"Created: {lead.name} ({lead.id})")

Async Client

import asyncio
from leadsdb import AsyncLeadsDB, Lead

async def main():
    async with AsyncLeadsDB("your-api-key") as client:
        lead = await client.create(Lead(
            name="Acme Corporation",
            source="website",
        ))
        print(f"Created: {lead.name}")

asyncio.run(main())

Client Options

client = LeadsDB(
    api_key,
    base_url="https://custom.api.com",
    timeout=30.0,
    max_retries=5,
)

CRUD Operations

Create

from leadsdb import Lead, text_attr, number_attr, bool_attr, list_attr

lead = client.create(Lead(
    name="Acme Corporation",
    source="website",
    description="Leading provider of innovative solutions",
    address="123 Main Street",
    city="San Francisco",
    state="CA",
    country="USA",
    postal_code="94105",
    latitude=37.7749,
    longitude=-122.4194,
    phone="[phone]",
    email="[e-mail]",
    website="https://acme.com",
    rating=4.5,
    review_count=128,
    category="Technology",
    tags=["enterprise", "saas", "b2b"],
    source_id="acme-001",
    logo_url="https://acme.com/logo.png",
    attributes=[
        text_attr("industry", "Software"),
        number_attr("employees", 500),
        bool_attr("verified", True),
        list_attr("products", ["CRM", "ERP", "Analytics"]),
    ],
))

Get

lead = client.get("lead-id")

Update

from leadsdb import UpdateLeadInput

lead = client.update("lead-id", UpdateLeadInput(
    rating=4.8,
    city="New York",
    tags=["enterprise", "saas", "b2b", "updated"],
))

Delete

client.delete("lead-id")

Listing Leads

Basic List

from leadsdb import city, rating, SortField, SortOrder

result = client.list(
    city().eq("Berlin"),
    rating().gte(4.0),
    sort_by=SortField.NAME,
    sort_order=SortOrder.ASC,
    limit=20,
)

for lead in result.leads:
    print(f"{lead.name} ({lead.city})")

# Pagination
if result.has_more:
    next_result = client.list(
        city().eq("Berlin"),
        cursor=result.next_cursor,
    )

Iterator (Automatic Pagination)

for lead in client.iterate(city().eq("Berlin"), sort_by=SortField.NAME):
    print(lead.name)

Async Iterator

async for lead in client.iterate(city().eq("Berlin")):
    print(lead.name)

Filters

All filters default to AND logic. Use or_() for OR logic.

Text Fields

Available for: name(), city(), country(), state(), category(), source(), email(), phone(), website()

Method	Description
eq(value)	Equals
neq(value)	Not equals
contains(value)	Contains substring
not_contains(value)	Does not contain substring
is_empty()	Field is empty
is_not_empty()	Field is not empty

from leadsdb import city, name, email

city().eq("Berlin")
name().contains("Tech")
email().is_not_empty()

Number Fields

Available for: rating(), review_count()

Method	Description
eq(value)	Equals
neq(value)	Not equals
gt(value)	Greater than
gte(value)	Greater than or equal
lt(value)	Less than
lte(value)	Less than or equal

from leadsdb import rating, review_count

rating().gte(4.0)
review_count().gt(100)

Array Fields

Available for: tags()

Method	Description
contains(value)	Array contains value
not_contains(value)	Array does not contain value
is_empty()	Array is empty
is_not_empty()	Array is not empty

from leadsdb import tags

tags().contains("enterprise")
tags().is_not_empty()

Location

Method	Description
within_radius(lat, lon, km)	Within radius in kilometers
is_set()	Coordinates are set
is_not_set()	Coordinates are not set

from leadsdb import location

# Find leads within 50km of Berlin
location().within_radius(52.52, 13.405, 50)

Custom Attributes

Use attr(name) for custom attribute filters:

from leadsdb import attr

attr("industry").eq("Software")
attr("employees").gte(100)

OR Logic

from leadsdb import or_, rating

# City is Berlin OR Paris
result = client.list(
    rating().gte(4.0),           # AND
    or_().city().eq("Berlin"),   # OR
    or_().city().eq("Paris"),    # OR
)

Sorting

from leadsdb import SortField, SortOrder, attr_sort_field

# Sort by field
result = client.list(sort_by=SortField.NAME, sort_order=SortOrder.ASC)
result = client.list(sort_by=SortField.RATING, sort_order=SortOrder.DESC)

# Sort by custom attribute
result = client.list(sort_by=attr_sort_field("employees"), sort_order=SortOrder.DESC)

Available sort fields:

• SortField.NAME, SortField.CITY, SortField.COUNTRY, SortField.STATE
• SortField.CATEGORY, SortField.SOURCE, SortField.EMAIL, SortField.PHONE, SortField.WEBSITE
• SortField.RATING, SortField.REVIEW_COUNT
• SortField.CREATED_AT, SortField.UPDATED_AT

Bulk Operations

Bulk Create (up to 100 leads)

result = client.bulk_create([
    Lead(name="Lead 1", source="import", city="Athens"),
    Lead(name="Lead 2", source="import", city="London"),
    Lead(name="Lead 3", source="import", city="Paris"),
])

print(f"Created: {result.success}, Failed: {result.failed}")
for created in result.created:
    print(f"ID: {created.id} (index {created.index})")

Notes

# Create a note
note = client.create_note("lead-id", "Initial contact made")

# List notes
notes = client.list_notes("lead-id")

# Update a note
note = client.update_note("note-id", "Updated content")

# Delete a note
client.delete_note("note-id")

Export

from leadsdb import ExportFormat

# Export as CSV
data = client.export(ExportFormat.CSV)
with open("leads.csv", "wb") as f:
    f.write(data)

# Export as JSON
data = client.export(ExportFormat.JSON)

Error Handling

from leadsdb import NotFoundError, APIError

try:
    lead = client.get("non-existent-id")
except NotFoundError:
    print("Lead not found")
except APIError as e:
    print(f"Status: {e.status_code}, Message: {e.message}")

Exception types:

• NotFoundError - Resource not found (404)
• UnauthorizedError - Invalid API key (401)
• ForbiddenError - Access denied (403)
• RateLimitedError - Too many requests (429)
• BadRequestError - Invalid request (400)
• ValidationError - Client-side validation error

Using ListOptions

For more complex queries, you can use ListOptions:

from leadsdb import ListOptions, city, rating, SortField, SortOrder

opts = (
    ListOptions()
    .with_limit(50)
    .with_sort(SortField.RATING, SortOrder.DESC)
    .with_filter(city().eq("Berlin"))
    .with_filter(rating().gte(4.0))
)

result = client.list(options=opts)

License

MIT