Python repository pattern implementation
Project description
repka
Python repository pattern implementation
Installation
Via pip:
pip install repka
Via poetry:
poetry add repka
Usage
from typing import Any
import sqlalchemy as sa
from aiopg.sa import create_engine
from repka.api import BaseRepository, IdModel
# Define SA table
metadata = sa.MetaData()
transactions_table = sa.Table(
"transactions",
metadata,
sa.Column("id", sa.Integer, primary_key=True, autoincrement=True),
...
)
# Define pydantic model
class Transaction(IdModel):
...
# Define repository
class TransactionRepo(BaseRepository):
table = transactions_table
def deserialize(self, **kwargs: Any) -> Transaction:
return Transaction(**kwargs)
# Create SA connection
connection_params = dict(user='aiopg', database='aiopg', host='127.0.0.1', password='passwd')
async with create_engine(**connection_params) as engine:
async with engine.acquire() as conn:
# Instantiate repository
repo = TransactionRepo(conn)
# Now you can use the repo
# Here we select first matching row from table and convert it to model
transaction = await repo.first(transactions_table.c.id == 1)
Tests
To run tests:
- Setup database url via
DATABASE_URL
environment variable
WARNING: Every test run will drop all tables from db
- Run tests via
pytest
Contribution
-
Create fork/branch for new feature/fix/whatever
-
Install pre-commit hooks:
pre-commit install
(for manual pre-commit run usepre-commit run -a
) -
When you done create pull request and wait for approval
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
repka-0.5.0.tar.gz
(4.0 kB
view hashes)
Built Distribution
repka-0.5.0-py3-none-any.whl
(8.7 kB
view hashes)