Skip to main content

One pattern for accessing data powered by SQLAlchemy & Pydantic.

Project description

OnePattern

One pattern for accessing data powered by SQLAlchemy & Pydantic.

Features

  • CRUD: Create, read, update, delete operations.
  • Pagination: Built-in support for pagination & sorting.
  • Validation: Automatic validation using Pydantic models.
  • Bulk operations: Create, update, delete multiple records at once.
  • Unit of work: Transactional support for multiple operations.

Requirements

OnePattern stands on the shoulders of giants:

Installation

pip install onepattern

Get Started

Let's write a simple CRUD API for managing users to demonstrate the power of OnePattern.

Create models:

from datetime import datetime

from pydantic import BaseModel, ConfigDict
from sqlalchemy import Identity
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Identity(), primary_key=True)
    name: Mapped[str]
    age: Mapped[int]
    salary: Mapped[int]
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    updated_at: Mapped[datetime] = mapped_column(
        default=datetime.now, onupdate=datetime.now
    )


class UserBase(BaseModel):
    name: str
    age: int
    salary: int

    model_config = ConfigDict(from_attributes=True)


class UserCreate(UserBase):
    pass


class UserRead(UserBase):
    id: int
    created_at: datetime
    updated_at: datetime

Create repository:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

from docs.gs_models import User, UserRead
from onepattern import AlchemyRepository


class UserRepository(AlchemyRepository[User, UserRead]):
    model_type = User
    schema_type = UserRead


async_engine = create_async_engine("sqlite+aiosqlite://", echo=True)
async_session = async_sessionmaker(async_engine)


async def get_users() -> UserRepository:
    async with async_session() as session:
        async with session.begin():
            yield UserRepository(session)

Use it in your app:

from contextlib import asynccontextmanager
from typing import Annotated, Any

from fastapi import FastAPI, Depends, HTTPException

from docs.gs_models import Base, UserCreate, UserRead
from docs.gs_repository import UserRepository, async_engine, get_user_repo
from onepattern import PageParams, Page


@asynccontextmanager
async def lifespan(_app: FastAPI) -> Any:
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/users/")
async def create_user(
        user: UserCreate, users: Annotated[UserRepository, Depends(get_user_repo)]
) -> UserRead:
    return await users.create(user)


async def get_user_dep(
        user_id: int, users: Annotated[UserRepository, Depends(get_user_repo)]
) -> UserRead:
    user = await users.get(user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@app.get("/users/{user_id}")
async def get_user(
        user: Annotated[UserRead, Depends(get_user_dep)],
) -> UserRead:
    return user


@app.put("/users/{user_id}")
async def update_user(
        update: UserCreate,
        user: Annotated[UserRead, Depends(get_user_dep)],
        users: Annotated[UserRepository, Depends(get_user_repo)],
) -> UserRead:
    return await users.update(user.id, update)


@app.delete("/users/{user_id}")
async def delete_user(
        user: Annotated[UserRead, Depends(get_user_dep)],
        users: Annotated[UserRepository, Depends(get_user_repo)],
) -> UserRead:
    return await users.delete(user.id)


@app.get("/users/")
async def get_users(
        params: Annotated[PageParams, Depends()],
        users: Annotated[UserRepository, Depends(get_user_repo)],
) -> Page[UserRead]:
    return await users.get_many(params=params)

Run app:

uvicorn docs.gs_app:app --host 0.0.0.0 --port 8000 --reload

img.png

Pagination example:

img.png img.png

Extending models

OnePattern provides multiple ways to extend models:

  1. Use mixins to add commonly-used columns:
from sqlalchemy.orm import DeclarativeBase, Mapped

from onepattern.models import HasID, HasTimestamp


class Base(DeclarativeBase):
    pass


class UserMixins(Base, HasID, HasTimestamp):
    __tablename__ = "users"

    name: Mapped[str]
    age: Mapped[int]
    salary: Mapped[int]

Tip: See onepattern.schemas for similar pydantic mixins.

  1. Use pre-configured base model:
from datetime import datetime

from sqlalchemy import Identity
from sqlalchemy.orm import Mapped, mapped_column

from onepattern import AlchemyBase


class UserAlchemyBase(AlchemyBase):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Identity(), primary_key=True)
    name: Mapped[str]
    age: Mapped[int]
    salary: Mapped[int]
    created_at: Mapped[datetime] = mapped_column(default=datetime.now)
    updated_at: Mapped[datetime] = mapped_column(
        default=datetime.now, onupdate=datetime.now
    )
  1. Use entity model with commonly-used columns:
from sqlalchemy.orm import Mapped

from onepattern import AlchemyEntity


class UserAlchemyEntity(AlchemyEntity):
    __tablename__ = "users"

    name: Mapped[str]
    age: Mapped[int]
    salary: Mapped[int]
    # id, created_at and updated_at are added automatically

Tip: See onepattern.schemas.EntityModel for similar pydantic base.

Made with love ❤️

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

onepattern-0.2.0.tar.gz (11.1 kB view details)

Uploaded Source

Built Distribution

onepattern-0.2.0-py3-none-any.whl (15.6 kB view details)

Uploaded Python 3

File details

Details for the file onepattern-0.2.0.tar.gz.

File metadata

  • Download URL: onepattern-0.2.0.tar.gz
  • Upload date:
  • Size: 11.1 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.8.2 CPython/3.12.3 Windows/11

File hashes

Hashes for onepattern-0.2.0.tar.gz
Algorithm Hash digest
SHA256 69fede41c8939a0f21a0914b8eaa2a2b534853d59847d3903e2335105645ef02
MD5 84be1eaedf1f8b23cc51bd58cfa40a2d
BLAKE2b-256 02aa695cac180710f6b7130b652758422ca2359a586274b9b34c026e65540a6e

See more details on using hashes here.

File details

Details for the file onepattern-0.2.0-py3-none-any.whl.

File metadata

  • Download URL: onepattern-0.2.0-py3-none-any.whl
  • Upload date:
  • Size: 15.6 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.8.2 CPython/3.12.3 Windows/11

File hashes

Hashes for onepattern-0.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 425445532bb94fe794753677427385d12823b92b5c7e21b3893f903b1e1f0486
MD5 86df420f1f1d6711249d0c68e1235386
BLAKE2b-256 3fec0d9eb25f08ee65878bf3b0267a281bc91d3f4d3b2f6a7456b63add2b804d

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page