Pynocular

Pynocular is a lightweight ORM that lets you query your database using Pydantic models and asyncio.

With Pynocular you can decorate 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
• 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
• Advanced transaction management system allows you to conditionally put requests in transactions

Table of Contents:

• Installation
• Guide
  • Basic Usage
  • Advanced Usage
  • Creating database tables
• Development

Installation

Pynocular requires Python 3.6 or above.

pip install pynocular
# or
poetry add pynocular

Guide

Basic Usage

Pynocular works by decorating your base Pydantic model with the function database_model. Once decorated with the proper information, you can proceed to use that model to interface with your specified database table.

The first step is to define a DBInfo object. This will contain the connection information to your database.

from pynocular.engines import DatabaseType, DBInfo


# 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"
)
db_info = DBInfo(connection_string)

Object Management

Once you define a db_info object, you are ready to decorate your Pydantic models and interact with your database!

from pydantic import BaseModel, Field
from pynocular.database_model import database_model, UUID_STR

from my_package import db_info

@database_model("organizations", db_info)
class Org(BaseModel):

    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)

#### Object management

# 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

DatabaseModels have their own serialization functions to convert to and from dictionaries.

# Serializing org with `to_dict()`
org = 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

Using Nested DatabaseModels

Pynocular also supports basic object relationships. If your database tables have a foreign key reference you can leverage that in your pydantic models to increase the accessibility of those related objects.

from pydantic import BaseModel, Field
from pynocular.database_model import database_model, nested_model, UUID_STR

from my_package import db_info

@database_model("users", db_info)
class User(BaseModel):

    id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True)
    username: str = Field(max_length=100)

    created_at: Optional[datetime] = Field(fetch_on_create=True)
    updated_at: Optional[datetime] = Field(fetch_on_update=True)

@database_model("organizations", db_info)
class Org(BaseModel):

    id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True)
    name: str = Field(max_length=45)
    slug: str = Field(max_length=45)
    # `organizations`.`tech_owner_id` is a foreign key to `users`.`id`
    tech_owner: Optional[nested_model(User, reference_field="tech_owner_id")]
    # `organizations`.`business_owner_id` is a foreign key to `users`.`id`
    business_owner: nested_model(User, reference_field="business_owner_id")
    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)


tech_owner = await User.create("tech owner")
business_owner = await User.create("business owner")


# Creating org with only business owner set
org = await Org.create(
    name="org name",
    slug="org-slug",
    business_owner=business_owner
)

assert org.business_owner == business_owner

# Add tech owner
org.tech_owner = tech_owner
await org.save()

# Fetch from the db and check ids
org2 = Org.get(org.id)
assert org2.tech_owner.id == tech_owner.id
assert org2.business_owner.id == business_owner.id

# Swap user roles
org2.tech_owner = business_owner
org2.business_owner = tech_owner
await org2.save()
org3 = await Org.get(org2.id)
assert org3.tech_owner.id == business_owner.id
assert org3.business_owner.id == tech_owner.id


# Serialize org
org_dict = org3.to_dict()
expected_org_dict = {
    "id": org3.id,
    "name": "org name",
    "slug": "org-slug",
    "business_owner_id": tech_owner.id,
    "tech_owner_id": business_owner.id,
    "tag": None,
    "created_at": org3.created_at,
    "updated_at": org3.updated_at
}

assert org_dict == expected_org_dict

When using DatabaseModel.get(..), any foreign references will need to be resolved before any properties besides the primary ID can be accessed. If you try to access a property before calling fetch() on the nested model, a NestedDatabaseModelNotResolved error will be thrown.

org_get = await Org.get(org3.id)
org_get.tech_owner.id # Does not raise `NestedDatabaseModelNotResolved`
org_get.tech_owner.username # Raises `NestedDatabaseModelNotResolved`

org_get = await Org.get(org3.id)
await org_get.tech_owner.fetch()
org_get.tech_owner.username # Does not raise `NestedDatabaseModelNotResolved`

Alternatively, calling DatabaseModel.get_with_refs() instead of DatabaseModel.get() will automatically fetch the referenced records and fully resolve those objects for you.

org_get_with_refs = await Org.get_with_refs(org3.id)
org_get_with_refs.tech_owner.username # Does not raise `NestedDatabaseModelNotResolved`

There are some situations where none of the objects have been persisted to the database yet. In this situation, you can call Database.save(include_nested_models=True) on the object with the references and it will persist all of them in a transaction.

# We create the objects but dont persist them
tech_owner = User("tech owner")
business_owner = User("business owner")

org = Org(
    name="org name",
    slug="org-slug",
    business_owner=business_owner
)

await org.save(include_nested_models=True)

Special Type arguments

With Pynocular you can set fields to be optional and set by the database. 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 or fetch_on_update=True to the Field 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 BaseModel, Field
from pynocular.database_model import database_model, nested_model, UUID_STR

from my_package import db_info

@database_model("user_subscriptions", db_info)
class UserSubscriptions(BaseModel):

    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 insert a bunch of records into a database and you don't want to do an insert for each on