Lightweight ORM that lets you query your database using Pydantic models and asyncio
Project description
Pynocular
Pynocular is a lightweight ORM that lets you query your database using Pydantic models and asyncio.
With Pynocular, you can annotate your existing Pydantic models to sync them with the corresponding table in your database, allowing you to persist changes without ever having to think about the database. Transaction management is automatically handled for you so you can focus on the important parts of your code. This integrates seamlessly with frameworks that use Pydantic models such as FastAPI.
Features:
- Fully supports asyncio to write to SQL databases through the databases library
- Provides simple methods for basic SQLAlchemy support (create, delete, update, read)
- Contains access to more advanced functionality such as custom SQLAlchemy selects
- Contains helper functions for creating new database tables
- Supports automatic and nested transactions
Table of Contents:
Installation
Pynocular requires Python 3.9 or above.
pip install pynocular
# or
poetry add pynocular
Basic Usage
Defining models
Pynocular works by augmenting Pydantic's BaseModel
through the DatabaseModel
class. Once you define a class that extends DatabaseModel
, you can proceed to use that model to interface with your specified database table.
from pydantic import Field
from pynocular import DatabaseModel, UUID_STR
class Org(DatabaseModel, table_name="organizations"):
id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True)
name: str = Field(max_length=45)
slug: str = Field(max_length=45)
tag: Optional[str] = Field(max_length=100)
created_at: Optional[datetime] = Field(fetch_on_create=True)
updated_at: Optional[datetime] = Field(fetch_on_update=True)
Creating a database and setting the backend
The first step is to create a database pool and set the Pynocular backend. This will tell the models how to persist data.
Use the databases library to create a database connection using the dialect of your choice and pass the database object to SQLDatabaseModelBackend
.
from pynocular import Database, set_backend, SQLDatabaseModelBackend
async def main():
# Example below shows how to connect to a locally-running Postgres database
connection_string = f"postgresql://{db_user_name}:{db_user_password}@localhost:5432/{db_name}?sslmode=disable"
async with Database(connection_string) as db:
with set_backend(SQLDatabaseModelBackend(db)):
print(await Org.get_list())
Creating, reading, updating, and deleting database objects
Once you define a database model and set a backend, you are ready to interact with your database!
# Create a new Org via `create`
org = await Org.create(name="new org", slug="new-org")
# Create a new Org via `save`
org2 = Org(name="new org2", slug="new-org2")
await org2.save()
# Update an org
org.name = "renamed org"
await org.save()
# Delete org
await org.delete()
# Get org
org3 = await Org.get(org2.id)
assert org3 == org2
# Get a list of orgs
orgs = await Org.get_list()
# Get a filtered list of orgs
orgs = await Org.get_list(tag="green")
# Get orgs that have several different tags
orgs = await Org.get_list(tag=["green", "blue", "red"])
# Fetch the latest state of a table in the db
org3.name = "fake name"
await org3.fetch()
assert org3.name == "new org2"
Serialization
Database models have their own serialization functions to convert to and from dictionaries.
# Serializing org with `to_dict()`
org = await Org.create(name="org serialize", slug="org-serialize")
org_dict = org.to_dict()
expected_org_dict = {
"id": "e64f6c7a-1bd1-4169-b482-189bd3598079",
"name": "org serialize",
"slug": "org-serialize",
"created_at": "2018-01-01 7:03:45",
"updated_at": "2018-01-01 9:24:12"
}
assert org_dict == expected_org_dict
# De-serializing org with `from_dict()`
new_org = Org.from_dict(expected_org_dict)
assert org == new_org
Special type arguments
With Pynocular you can set fields to be optional and rely on the database server to set its value. This is useful
if you want to let the database autogenerate your primary key or created_at
and updated_at
fields
on your table. To do this you must:
- Wrap the typehint in
Optional
- Provide keyword arguments of
fetch_on_create=True
orfetch_on_update=True
to theField
class
Advanced Usage
For most use cases, the basic usage defined above should suffice. However, there are certain situations
where you don't necessarily want to fetch each object or you need to do more complex queries that
are not exposed by the DatabaseModel
interface. Below are some examples of how those situations can
be addressed using Pynocular.
Tables with compound keys
Pynocular supports tables that use multiple fields as its primary key such as join tables.
from pydantic import Field
from pynocular import DatabaseModel, UUID_STR
class UserSubscriptions(DatabaseModel, table_name="user_subscriptions"):
user_id: UUID_STR = Field(primary_key=True, fetch_on_create=True)
subscription_id: UUID_STR = Field(primary_key=True, fetch_on_create=True)
name: str
user_sub = await UserSub.create(
user_id="4d4254c4-8e99-45f9-8261-82f87991c659",
subscription_id="3cc5d476-dbe6-4cc1-9390-49ebd7593a3d",
name="User 1's subscriptions"
)
# Get the users subscription and confirm its the same
user_sub_get = await UserSub.get(
user_id="4d4254c4-8e99-45f9-8261-82f87991c659",
subscription_id="3cc5d476-dbe6-4cc1-9390-49ebd7593a3d",
)
assert user_sub_get == user_sub
# Change a property value like any other object
user_sub_get.name = "change name"
await user_sub_get.save()
Batch operations on tables
Sometimes you want to perform a bulk insert of records into a database table.
This can be handled by the create_list
function.
org_list = [
Org(name="org1", slug="org-slug1"),
Org(name="org2", slug="org-slug2"),
Org(name="org3", slug="org-slug3"),
]
await Org.create_list(org_list)
This function will insert all records into your database table in one batch.
If you have a use case that requires deleting a bunch of records based on some field value, you can use delete_records
:
# Delete all records with the tag "green"
await Org.delete_records(tag="green")
# Delete all records with if their tag has any of the following: "green", "blue", "red"
await Org.delete_records(tag=["green", "blue", "red"])
Sometimes you may want to update the value of a record in a database without having to fetch it first. This can be accomplished by using
the update_record
function:
await Org.update_record(
id="05c0060c-ceb8-40f0-8faa-dfb91266a6cf",
tag="blue"
)
org = await Org.get("05c0060c-ceb8-40f0-8faa-dfb91266a6cf")
assert org.tag == "blue"
Transactions and asyncio.gather
You should avoid using asyncio.gather
within a database transaction. You can use Pynocular's gather
function instead, which has the same interface but executes queries sequentially:
from pynocular import get_backend
from pynocular.util import gather
async with get_backend().transaction():
await gather(
Org.create(id="abc", name="foo"),
Org.create(id="def", name="bar"),
)
The reason is that concurrent queries can interfere with each other and result in the error:
asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress
See: https://github.com/encode/databases/issues/125#issuecomment-511720013
Complex queries
Sometimes your application will require performing complex queries, such as getting the count of each unique field value for all records in the table. Because Pynocular is backed by SQLAlchemy, we can access table columns directly to write pure SQLAlchemy queries as well!
from sqlalchemy import func, select
from pynocular import get_backend
async def generate_org_stats():
query = (
select([func.count(Org.column.id), Org.column.tag])
.group_by(Org.column.tag)
.order_by(func.count().desc())
)
# Get the active backend and open a database transaction
async with get_backend().transaction():
result = await conn.execute(query)
return [dict(row) for row in result]
Creating database and tables
With Pynocular you can use simple Python code to create new databases and database tables. All you need is a working connection string to the database host and a properly defined DatabaseModel
class. When you define a class that extends DatabaseModel
, Pynocular creates a SQLAlchemy table under the hood. This can be accessed via the table
property.
from pynocular import Database
from pynocular.util import create_new_database, create_table
from my_package import Org
async def main():
connection_string = "postgresql://postgres:XXXX@localhost:5432/postgres"
await create_new_database(connection_string, "my_new_db")
connection_string = "postgresql://postgres:XXXX@localhost:5432/my_new_db"
async with Database(connection_string) as db:
# Creates a new database and "organizations" table in that database
await create_table(db, Org.table)
Unit testing with DatabaseModel
Pynocular comes with tooling to write unit tests against your database models, giving you
the ability to test your business logic without the extra work and latency involved in
managing a database. All you have to do is set the backend using the MemoryDatabaseModelBackend
instead of the SQL backend. You don't need to change any of your database model definitions.
from pynocular import MemoryDatabaseModelBackend, set_backend
from my_package import Org, User
async def main():
orgs = [
Org(id=str(uuid4()), name="orgus borgus", slug="orgus_borgus"),
Org(id=str(uuid4()), name="orgus borgus2", slug="orgus_borgus"),
]
with set_backend(MemoryDatabaseModelBackend()):
await Org.create_list(orgs)
fetched_orgs = await Org.get_list(name=orgs[0].name)
assert orgs[0] == fetched_orgs[0]
users = [
User(id=str(uuid4()), username="Bob"),
User(id=str(uuid4()), username="Sally"),
]
# You can also seed the backend with existing records
with MemoryDatabaseModelBackend(
records={
"orgs": [o.to_dict() for o in orgs],
"users": [u.to_dict() for u in users],
}
):
org = await Org.get(orgs[0].id)
org.name = "new test name"
await org.save()
Development
To develop Pynocular, install dependencies and enable the pre-commit hook. Make sure to install Python 3.9 and activate it in your shell.
sudo yum install libffi-devel # Needed for ctypes to install poetry
pyenv install 3.9.12
pyenv shell 3.9.12
Install dependencies and enable the pre-commit hook.
pip install pre-commit poetry
poetry install
pre-commit install
Run tests to confirm everything is installed correctly.
poetry run pytest
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for pynocular-2.0.0rc3-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | a2ec70736a731f7fc6d79eb38a1f833ba91bd993095830e57e81332f6cd03175 |
|
MD5 | 049f0ef2f42334ee945ae3898b66217c |
|
BLAKE2b-256 | 9713a625fadfa4fb058e60f9711d56df472a8d4d028d2953723f3b0c41c74a5e |