Piccolo ORM
Piccolo is a fast, user-friendly, and fully type-annotated ORM and query builder for Python, primarily focused on asynchronous operations. It supports PostgreSQL, SQLite, and CockroachDB, and comes with batteries included like migrations, an admin GUI, and ASGI application templates. Currently at version 1.33.0, Piccolo releases frequently, often with minor version bumps every few weeks, incorporating new features and improvements.
Common errors
-
ModuleNotFoundError: No module named 'piccolo.engine.postgres'
cause The required database driver for PostgreSQL (asyncpg) or SQLite (aiosqlite) is not installed.fixInstall the necessary driver: `pip install "piccolo[postgres]"` for PostgreSQL or `pip install "piccolo[sqlite]"` for SQLite. -
RuntimeWarning: No database engine found. Make sure 'DB' is defined in your `piccolo_conf.py` or passed directly to your Tables.
cause Piccolo could not find a configured database engine, often because `piccolo_conf.py` is missing, incorrectly configured, or the engine isn't explicitly passed to `Table` instances.fixDefine a `piccolo_conf.py` file in your project root with a `DB` variable pointing to your `Engine` instance, or pass the `db=YOUR_ENGINE` argument directly to your `Table` classes. Ensure the engine is initialized (e.g., `await DB.start_connection_pool()`). -
sqlite3.OperationalError: database is locked
cause This typically occurs with SQLite when multiple asynchronous operations attempt to write to the database concurrently, or when a connection isn't properly closed.fixEnsure `DB.close_connection_pool()` is called after all operations complete. For heavy concurrent workloads, consider using PostgreSQL or CockroachDB, which handle concurrency better than SQLite. -
AttributeError: type object 'Band' has no attribute 'objects'
cause In modern Piccolo versions, queries are typically run directly on the `Table` class or instances, not through a separate `objects()` manager, though `Table.objects()` style queries are shown in older examples or for specific use cases.fixRewrite your query to use `await Band.select().run()`, `await Band.insert().run()`, etc., directly on the `Table` class. The `.objects()` call might be relevant for specific patterns like `Band.objects().get(...)` but isn't the primary query builder interface.
Warnings
- breaking Using UUID v7 for columns requires Python 3.14 and PostgreSQL 18. Attempting to use this feature on older versions may result in errors or unexpected behavior.
- gotcha When defining columns, ensure you use `null=True` for nullable fields. Piccolo includes typo detection and will warn if `nul=True` is used instead.
- deprecated The `graphlib` backport was removed in version 1.31.0 as it's no longer needed in supported Python versions. This mostly affects internal dependencies, but extremely old Python versions relying on the backport might behave unexpectedly.
- gotcha A bug existed in auto migrations where `ForeignKey` columns specifying `target_column` could lead to multiple primary key columns being added. This was fixed in 1.26.1.
- gotcha When adding a new column to an existing table via auto migrations, it must initially be set to `null=True`. You can make it non-nullable in a subsequent migration.
Install
-
pip install piccolo -
pip install "piccolo[all]" -
pip install "piccolo[postgres]"
Imports
- Table
from piccolo.table import Table
- Varchar
from piccolo.columns import Varchar
- Integer
from piccolo.columns import Integer
- SQLiteEngine
from piccolo.engine.sqlite import SQLiteEngine
- PostgresEngine
from piccolo.engine.postgres import PostgresEngine
Quickstart
import asyncio
from piccolo.table import Table
from piccolo.columns import Varchar, Integer
from piccolo.engine.sqlite import SQLiteEngine
# 1. Define your database engine (in-memory SQLite for quick start)
DB = SQLiteEngine(path=':memory:')
# 2. Define your Table
class Band(Table, db=DB):
name = Varchar(length=100)
popularity = Integer(default=0)
async def main():
# 3. Create tables
await Band.create_table(if_not_exists=True)
# 4. Insert data
await Band.insert(
Band(name="Pythonistas", popularity=1000),
Band(name="Asyncio Allstars", popularity=800)
).run()
# 5. Select data
all_bands = await Band.select(Band.name, Band.popularity).run()
print("All bands:", all_bands)
popular_bands = await Band.select(Band.name).where(Band.popularity > 900).run()
print("Popular bands:", popular_bands)
# 6. Update data
await Band.update({"popularity": 1100}).where(Band.name == "Pythonistas").run()
updated_bands = await Band.select(Band.name, Band.popularity).run()
print("Updated bands:", updated_bands)
# 7. Close the connection pool (important for persistent databases)
await DB.close_connection_pool()
if __name__ == "__main__":
asyncio.run(main())