🦆 Duckling

A Beanie-inspired ORM for DuckDB — async-first, Pydantic-powered.

Duckling brings the elegant, developer-friendly API of Beanie (MongoDB ODM) to DuckDB — the fast, in-process analytical database. Define your models with Pydantic, query with Pythonic expressions, and enjoy both async and sync APIs.

The original version of this project was generated by Claude AI.

The human intervention in the code is to customize it for use within the carbonbits ecosystem. No plans to publish to pypi yet

Installation

pip install duckling
# or from source:
pip install -e .

Requirements: Python ≥ 3.10, duckdb >= 0.9, pydantic >= 2.0

---

Quick Start

importasynciofromtypingimportAnnotated, OptionalfromducklingimportDocument, IndexSpec, init_ducklingclassUser(Document):
name: stremail: Annotated[str, IndexSpec(unique=True)]
age: int=0classSettings:
table_name="users"asyncdefmain():
awaitinit_duckling(database=":memory:", document_models=[User])
# Insertalice=User(name="Alice", email="alice@example.com", age=30)
awaitalice.insert()
# Queryusers=awaitUser.find(User.age>25).sort("+name").limit(10).to_list()
# Updatealice.age=31awaitalice.save()
# Deleteawaitalice.delete()
asyncio.run(main())

---

API Reference

Initialization

fromducklingimportinit_duckling, init_duckling_sync# Asyncawaitinit_duckling(
database=":memory:", # or "path/to/file.db"document_models=[User, Product],
recreate_tables=False, # drop & recreate tables
)
# Syncinit_duckling_sync(database="app.db", document_models=[User])

Defining Models

Duckling models are Pydantic BaseModel subclasses. By default, id is an auto-generated, auto-incrementing integer primary key:

fromducklingimportDocument, IndexSpecfromtypingimportAnnotated, Optional, ListimportdatetimeclassProduct(Document):
name: strprice: floatcategory: Optional[str] =Nonetags: Optional[List[str]] =None# stored as JSONcreated_at: datetime.datetime=datetime.datetime.now()
in_stock: bool=TrueclassSettings:
table_name="products"# optional, auto-generated from class name

Supported types:str, int, float, bool, bytes, datetime.date, datetime.datetime, datetime.time, uuid.UUID, Optional[T], List[T] (→ JSON), dict (→ JSON), nested Pydantic models (→ JSON), Enum.

Custom Primary Keys (UUID, ULID, string)

Override id with any type other than int to opt out of the auto-increment sequence and use your own primary key — a UUID, a sortable ULID, or a plain caller-supplied string:

importuuidfrompydanticimportFieldfromducklingimportDocument, generate_ulidclassSession(Document):
id: str=Field(default_factory=generate_ulid) # sortable string PKuser_email: strclassApiKey(Document):
id: uuid.UUID=Field(default_factory=uuid.uuid4) # UUID PKlabel: strclassTag(Document):
id: str# no default — caller must supply an idname: str

Inserting a document whose id already exists raises DocumentAlreadyExists.

Indexed Fields

fromducklingimportIndexSpecfromtypingimportAnnotatedclassUser(Document):
email: Annotated[str, IndexSpec(unique=True)] # unique indexage: Annotated[int, IndexSpec()] # regular index

CRUD Operations

Every method has an async version (default) and a _sync variant:

Async	Sync	Description
await doc.insert()	doc.insert_sync()	Insert a new document
await doc.save()	doc.save_sync()	Upsert (insert or update)
await doc.delete()	doc.delete_sync()	Delete this document
await doc.refresh()	—	Reload from database
await Model.insert_many([...])	Model.insert_many_sync([...])	Bulk insert
await Model.delete_all()	Model.delete_all_sync()	Delete all rows
await Model.get(id)	Model.get_sync(id)	Fetch by primary key
await Model.count()	Model.count_sync()	Count all rows

Queries

Duckling's query interface mirrors Beanie's fluent API:

# Find with conditionsusers=awaitUser.find(User.age>25).to_list()
users=awaitUser.find(User.age>25, User.active==True).to_list()
# Find oneuser=awaitUser.find_one(User.email=="alice@example.com")
# Find allall_users=awaitUser.find_all().to_list()
# Chainingresults= (
awaitUser.find(User.active==True)
.find(User.age>=18) # additional conditions (AND)
.sort("+name") # ascending
.sort("-age") # descending
.skip(10) # offset
.limit(20) # limit
.to_list()
)
# Count & existscount=awaitUser.find(User.age>30).count()
has_any=awaitUser.find(User.name=="Alice").exists()
# Async iterationasyncforuserinUser.find(User.active==True).sort("+name"):
print(user.name)

Query Expressions

Use Pythonic operators directly on model fields:

# Comparison operatorsUser.age==30User.age!=30User.age>25User.age>=25User.age<40User.age<=40# Boolean combinators
(User.age>25) & (User.active==True) # AND
(User.name=="A") | (User.name=="B") # OR~(User.active==True) # NOT# FieldProxy helper methodsUser.name.startswith("Ali") # LIKE 'Ali%'User.name.endswith("son") # LIKE '%son'User.name.contains("lic") # LIKE '%lic%'User.name.like("A%e") # LIKE 'A%e'User.name.ilike("alice") # ILIKE (case-insensitive)User.age.is_in([25, 30, 35]) # IN (25, 30, 35)User.age.not_in([0, 99]) # NOT INUser.age.between(18, 65) # BETWEEN 18 AND 65# Sort helpersUser.name.asc() # → ("name", ASCENDING)User.name.desc() # → ("name", DESCENDING)

Operator Functions

For more complex queries, use the operator functions:

fromduckling.operatorsimportAnd, Or, Not, In, NotIn, Between, Like, ILike, RawawaitUser.find(In(User.age, [25, 30, 35])).to_list()
awaitUser.find(Between(User.age, 18, 65)).to_list()
awaitUser.find(Like(User.name, "%smith%")).to_list()
# CombineawaitUser.find(
And(
User.active==True,
Or(User.city=="NYC", User.city=="LA"),
Not(User.age<18),
)
).to_list()
# Raw SQL escape hatchawaitUser.find(Raw('"age" % 2 = 0')).to_list()

Aggregation

fromduckling.queryimportCount, Sum, Avg, Min, Max, CountDistinctstats=awaitUser.find(User.active==True).aggregate(
total=Count(),
avg_age=Avg("age"),
max_age=Max("age"),
min_age=Min("age"),
sum_age=Sum("age"),
unique_names=CountDistinct("name"),
)
print(stats) # {'total': 42, 'avg_age': 31.5, ...}

Sort Syntax

# String syntax
.sort("+name") # ascending
.sort("-age") # descending
.sort("+name", "-age") # multi-column# Tuple syntax
.sort(("name", SortDirection.ASCENDING))
# FieldProxy syntax
.sort(User.name.asc(), User.age.desc())

Transactions

session=get_session()
# Asyncasyncwithsession.async_transaction():
awaituser.insert()
awaitorder.insert()
# Syncwithsession.transaction():
user.insert_sync()
order.insert_sync()

Raw SQL Escape Hatch

fromducklingimportget_sessionsession=get_session()
# Asyncrows=awaitsession.async_fetchall("SELECT * FROM users WHERE age > ?", [25])
# Get pandas DataFramedf=awaitsession.async_fetchdf("SELECT name, age FROM users")
# Syncrows=session.fetchall("SELECT count(*) FROM users")

---

Beanie → Duckling Comparison

Beanie (MongoDB)	Duckling (DuckDB)
init_beanie(database, models)	await init_duckling(database, models)
class User(Document)	class User(Document)
await user.insert()	await user.insert()
await user.save()	await user.save()
await User.find(cond).to_list()	await User.find(cond).to_list()
await User.find_one(cond)	await User.find_one(cond)
User.name == "Alice"	User.name == "Alice"
In(User.age, [...])	In(User.age, [...])
await User.find().sort("+name")	await User.find().sort("+name")
Settings class	Settings class
Indexed(str, unique=True)	Annotated[str, IndexSpec(unique=True)]

---

Project Structure

src/duckling/
├── __init__.py # Public exports
├── connection.py # DuckDB session management
├── document.py # Document base class (the core)
├── fields.py # FieldProxy, Indexed, Expression types
├── init.py # init_duckling() / init_duckling_sync()
├── operators.py # And, Or, In, Between, Like, etc.
├── query.py # FindQuery builder + aggregation
└── exceptions.py # Custom exceptions
tests/ # Per-module tests (test_document.py, test_query.py, …)

License

MIT