SDKs

Copy page

SDKs

Official SDKs for integrating LeadsDB into your applications.

Go SDK

The official Go SDK provides a type-safe client for the LeadsDB API with automatic pagination, bulk operations, and comprehensive error handling.

Installation

Code
 
go get github.com/gosom/go-leadsdb

Requires Go 1.23+.

Quick Start

Code
 
package main

import (
    "context"
    "fmt"

    "github.com/gosom/go-leadsdb"
)

func main() {
    client := leadsdb.New("your-api-key")
    ctx := context.Background()

    // Create a lead
    lead, err := client.Create(ctx, &leadsdb.Lead{
        Name:    "Acme Corporation",
        Source:  "website",
        City:    "San Francisco",
        Email:   "[email protected]",
        Rating:  leadsdb.Ptr(4.5),
        Tags:    []string{"enterprise", "saas"},
    })
    if err != nil {
        panic(err)
    }

    fmt.Printf("Created: %s (%s)\n", lead.Name, lead.ID)
}

Client Options

Code
 
client := leadsdb.New(apiKey,
    leadsdb.WithBaseURL("https://custom.api.com"),
    leadsdb.WithTimeout(30 * time.Second),
    leadsdb.WithMaxRetries(5),
    leadsdb.WithHTTPClient(customClient),
)

CRUD Operations

Create

Code
 
lead, err := client.Create(ctx, &leadsdb.Lead{
    Name:        "Acme Corporation",
    Source:      "website",
    Description: "Leading provider of innovative solutions",
    Address:     "123 Main Street",
    City:        "San Francisco",
    State:       "CA",
    Country:     "USA",
    PostalCode:  "94105",
    Coordinates: &leadsdb.Coordinate{
        Latitude:  37.7749,
        Longitude: -122.4194,
    },
    Phone:       "+1-555-123-4567",
    Email:       "[email protected]",
    Website:     "https://acme.com",
    Rating:      leadsdb.Ptr(4.5),
    ReviewCount: leadsdb.Ptr(128),
    Category:    "Technology",
    Tags:        []string{"enterprise", "saas", "b2b"},
    SourceID:    "acme-001",
    LogoURL:     "https://acme.com/logo.png",
    Attributes: []leadsdb.Attribute{
        leadsdb.TextAttr("industry", "Software"),
        leadsdb.NumberAttr("employees", 500),
        leadsdb.BoolAttr("verified", true),
        leadsdb.ListAttr("products", []string{"CRM", "ERP", "Analytics"}),
        leadsdb.ObjectAttr("social", map[string]any{
            "linkedin": "https://linkedin.com/company/acme",
            "twitter":  "https://twitter.com/acme",
        }),
    },
})

Get

Code
 
lead, err := client.Get(ctx, "lead-id")

Update

Use UpdateLeadInput with pointer fields for partial updates:

Code
 
lead, err := client.Update(ctx, "lead-id", &leadsdb.UpdateLeadInput{
    Rating: leadsdb.Ptr(4.8),
    City:   leadsdb.Ptr("New York"),
    Tags:   []string{"enterprise", "saas", "b2b", "updated"},
})

Delete

Code
 
err := client.Delete(ctx, "lead-id")

Listing Leads

Basic List

Code
 
result, err := client.List(ctx,
    leadsdb.City().Eq("Berlin"),
    leadsdb.Rating().Gte(4.0),
    leadsdb.Sort(leadsdb.FieldName, leadsdb.Asc),
    leadsdb.Limit(20),
)

for _, lead := range result.Leads {
    fmt.Printf("%s (%s)\n", lead.Name, lead.City)
}

// Pagination
if result.HasMore {
    nextResult, err := client.List(ctx,
        leadsdb.City().Eq("Berlin"),
        leadsdb.Cursor(result.NextCursor),
    )
}

Iterator (Automatic Pagination)

Code
 
for lead, err := range client.Iterator(ctx,
    leadsdb.City().Eq("Berlin"),
    leadsdb.Sort(leadsdb.FieldName, leadsdb.Desc),
) {
    if err != nil {
        panic(err)
    }
    fmt.Printf("%s\n", lead.Name)
}

Channel-Based Iterator

Code
 
leads, errs := client.IteratorChan(ctx,
    leadsdb.Source().Eq("import"),
    leadsdb.Limit(100),
)

go func() {
    for err := range errs {
        log.Printf("Error: %v", err)
    }
}()

for lead := range leads {
    fmt.Printf("%s\n", 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
NotContains(value)	Does not contain substring
IsEmpty()	Field is empty
IsNotEmpty()	Field is not empty

Code
 
// Examples
leadsdb.City().Eq("Berlin")
leadsdb.Name().Contains("Tech")
leadsdb.Email().IsNotEmpty()

Number Fields

Available for: Rating(), ReviewCount()

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

Code
 
// Examples
leadsdb.Rating().Gte(4.0)
leadsdb.ReviewCount().Gt(100)

Array Fields

Available for: Tags()

Method	Description
Contains(value)	Array contains value
NotContains(value)	Array does not contain value
IsEmpty()	Array is empty
IsNotEmpty()	Array is not empty

Code
 
// Examples
leadsdb.Tags().Contains("enterprise")
leadsdb.Tags().IsNotEmpty()

Location

Method	Description
WithinRadius(lat, lon, km)	Within radius in kilometers
IsSet()	Coordinates are set
IsNotSet()	Coordinates are not set

Code
 
// Find leads within 50km of Berlin
leadsdb.Location().WithinRadius(52.52, 13.405, 50)

Custom Attributes

Use Attr(name) for custom attribute filters:

Method	Description
Eq(value)	Text equals
Neq(value)	Text not equals
Contains(value)	Text contains
EqNumber(value)	Number equals
Gt(value)	Number greater than
Gte(value)	Number greater than or equal
Lt(value)	Number less than
Lte(value)	Number less than or equal

Code
 
// Examples
leadsdb.Attr("industry").Eq("Software")
leadsdb.Attr("employees").Gte(100)

OR Logic

Code
 
// City is Berlin OR Paris
leadsdb.Or().City().Eq("Berlin")
leadsdb.Or().City().Eq("Paris")

// Combined with AND
client.List(ctx,
    leadsdb.Rating().Gte(4.0),         // AND
    leadsdb.Or().City().Eq("Berlin"),  // OR
    leadsdb.Or().City().Eq("Paris"),   // OR
)

Sorting

Code
 
// Sort by field
leadsdb.Sort(leadsdb.FieldName, leadsdb.Asc)
leadsdb.Sort(leadsdb.FieldRating, leadsdb.Desc)
leadsdb.Sort(leadsdb.FieldCreatedAt, leadsdb.Desc)

// Sort by custom attribute
leadsdb.Sort(leadsdb.AttrSortField("employees"), leadsdb.Desc)

Available sort fields:

• FieldName, FieldCity, FieldCountry, FieldState
• FieldCategory, FieldSource, FieldEmail, FieldPhone, FieldWebsite
• FieldRating, FieldReviewCount
• FieldCreatedAt, FieldUpdatedAt

Bulk Operations

Bulk Create (up to 100 leads)

Code
 
result, err := client.BulkCreate(ctx, []*leadsdb.Lead{
    {Name: "Lead 1", Source: "import", City: "Athens"},
    {Name: "Lead 2", Source: "import", City: "London"},
    {Name: "Lead 3", Source: "import", City: "Paris"},
})

fmt.Printf("Created: %d, Failed: %d\n", result.Success, result.Failed)
for _, created := range result.Created {
    fmt.Printf("ID: %s (index %d)\n", created.ID, created.Index)
}

Bulk Create from Channel (Streaming)

Auto-batches leads and flushes when batch is full or after timeout:

Code
 
leads := make(chan *leadsdb.Lead)

go func() {
    defer close(leads)
    for i := range 1000 {
        leads <- &leadsdb.Lead{
            Name:   fmt.Sprintf("Lead %d", i),
            Source: "import",
        }
    }
}()

results, errs := client.BulkCreateFromChan(ctx, leads)

go func() {
    for err := range errs {
        log.Printf("Error: %v", err)
    }
}()

for result := range results {
    fmt.Printf("Created: %s\n", result.ID)
}

Notes

Code
 
// Create a note
note, err := client.CreateNote(ctx, "lead-id", "Initial contact made")

// List notes
notes, err := client.ListNotes(ctx, "lead-id")

// Update a note
note, err = client.UpdateNote(ctx, "note-id", "Updated content")

// Delete a note
err = client.DeleteNote(ctx, "note-id")

Export

Code
 
reader, err := client.Export(ctx, leadsdb.ExportCSV)
if err != nil {
    panic(err)
}
defer reader.Close()

// Stream to file
file, _ := os.Create("leads.csv")
io.Copy(file, reader)

Export formats: ExportCSV, ExportJSON, ExportXLSX

Error Handling

Code
 
lead, err := client.Get(ctx, "non-existent-id")
if errors.Is(err, leadsdb.ErrNotFound) {
    fmt.Println("Lead not found")
}

// API errors include details
var apiErr *leadsdb.APIError
if errors.As(err, &apiErr) {
    fmt.Printf("Status: %d, Message: %s\n", apiErr.StatusCode, apiErr.Message)
}

Sentinel errors:

• ErrNotFound - Resource not found (404)
• ErrUnauthorized - Invalid API key (401)
• ErrForbidden - Access denied (403)
• ErrRateLimited - Too many requests (429)
• ErrBadRequest - Invalid request (400)

More Information

• GitHub Repository
• GoDoc Reference

---

Python SDK

The official Python SDK provides sync and async clients with Pydantic models, fluent filter API, and automatic pagination.

Installation

Code
 
pip install leadsdb

Requires Python 3.10+.

Quick Start

Code
 
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="[email protected]",
    rating=4.5,
    tags=["enterprise", "saas"],
))

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

Async Client

Code
 
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

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

CRUD Operations

Create

Code
 
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="+1-555-123-4567",
    email="[email protected]",
    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

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

Update

Code
 
from leadsdb import UpdateLeadInput

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

Delete

Code
 
client.delete("lead-id")

Listing Leads

Basic List

Code
 
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)

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

Async Iterator

Code
 
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

Code
 
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

Code
 
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

Code
 
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

Code
 
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:

Code
 
from leadsdb import attr

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

OR Logic

Code
 
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

Code
 
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)

Code
 
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

Code
 
# 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

Code
 
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)

Export formats: ExportFormat.CSV, ExportFormat.JSON, ExportFormat.XLSX

Error Handling

Code
 
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:

Code
 
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)

More Information

• PyPI Package
• GitHub Repository