steerage 2.0.0

Pluggable async storage backends for Python projects

Steerage

Pluggable async storage backends for Python projects.

This project implements the Repository Pattern for data access, using Pydantic 2.0 models for entity composition and decomposition.

Features

• One model, many possible storage backends: Start simple and add complexity as you need it.
• Use a fast, ephemeral in-memory database for tests, and a slow, reliable SQL database for production.
• One unified query interface: write most queries once, apply them to any backend.
• Simple environment-based configuration with Convoke
• 100% test coverage

Installation

With Pip:

pip install steerage

Example

Let's create a simple blog engine. First, we'll want a Pydantic entity model:

from uuid import UUID
from datetime import datetime
from pydantic import BaseModel, Field
from pydantic.types import AwareDatetime

class Entry(BaseModel):
    id: UUID
    path: str
    title: str
    body: str

    created_at: AwareDatetime = Field(
        default_factory=lambda: datetime.utcnow().replace(tzinfo=timezone.utc)
    )
    published_at: AwareDatetime | None = Field(default=None)

Models are all fine and good, but we need to be able to store it somewhere. Let's start simple, with an in-memory repository and a repository factory:

from steerage.repositories.memdb import AbstractInMemoryRepository


class InMemoryEntryRepository(AbstractInMemoryRepository):
    table_name: str = "entries"
    entity_class = Entry


def get_entry_repository():
    return InMemoryEntityRepository()

Now, in the async request handler, we can easily create an entry:

async def create_entry(request: Request):
    form = await EntryForm.from_request(request)
    if form.is_valid():
        repo = get_entry_repository()
        entry = Entry.model_validate(form)
        await repo.insert(entry)
        return redirect('entry-admin')

... and retrieve an entry:

async def get_entry(request: Request, id: UUID):
    repo = get_entry_repository()
    entry = await repo.get(id)
    return render('entry.html', {'entry': entry})

This is great, but whenever we restart the server, we lose all our data!

Let's go a step up and add a more durable disk-based repository:

from convoke.configs import BaseConfig
from steerage.repositories.shelvedb import ShelveInMemoryRepository


class ShelveEntryRepository(AbstractShelveRepository):
    table_name: str = "entries"
    entity_class = Entry


def get_entry_repository():
    config = BaseConfig()
    if config.TESTING:
        return InMemoryEntityRepository()
    else:
        return ShelveEntityRepository()

Now we have a durable repository for development use, and an in-memory repository for our tests.

Contribute

• Source Code: https://github.com/eykd/steerage
• Issue Tracker: https://github.com/eykd/steerage/issues

License

The project is licensed under the BSD 3-clause license.