onepattern 0.2.6

Description:

onepattern 0.2.6

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:

SQLAlchemy
Pydantic

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


Pagination example:


Extending models
OnePattern provides multiple ways to extend models:

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.


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
)


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 ❤️