Pluggable async storage backends for Python projects
Project description
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.
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.