Integrates SQLModel with wintry framework
Project description
WINTRY INTEGRATION FOR SQLMODEL
These functionalities were originaly part of wintry, but were moved to a separate package so it is an opt-in. wintry is really extensible, so, more features like this will be added in the future.
What is this
wintry-models is just a way of ease the use of SQLModel with wintry. It is exploiting wintry's Dependency Injection System to provide Request Bound SqlAlchemy Session, so changes to entities can be tracked across multiple functions, and dependencies got declared in a declarative way.
It is doable to acomplish the same with pure FastAPI, but is not trivial, besides, the integration with "Controllers" is a real Nightmare. This way, we promote a separation of concerns and a way to easily integrate an ORM into a web framework (like happens .NET).
Usage
Register the DataContext service and wire up the engine creation
from wintry import App, scoped
from wintrymodels.dbcontext import DataContext, AsyncSQLEngineContext, add_data_context
from sqlmodel.ext.asyncio.session import AsyncSession
# Declare the AppDbContext
@scoped
class AppDbContext(AsyncSession, DataContext):
def __init__(self):
super().__init__(AsyncSQLEngineContext.get_client(), expire_on_commit=False)
app = App()
# Register the Context engine
add_data_context(app, AsyncSQLEngineContext, "sqlite:///file.db")
We can now access SQLModel funcionalities as usual and have dependency injection, so we can do as follows:
from wintry import controller, get
from sqlmodel import select, SQLModel
class User(SQLModel):
id: int | None = None
username: str
@controller
class HeroController(object):
context: AppDbContext
@get("/users")
async def get_users(self) -> list[User]:
return await self.context.exec(select(User))
A simple UnitOfWork implementation
As we can register our AppDbContext
as a scoped dependency, we can share it among some
classes, so we can implement a UnitOfWork as follows
from wintry import scoped, controller, post, Path
from wintrymodels.dbcontext import DataContext, AsyncSQLEngineContext
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import SQLModel
from contextlib import asynccontextmanager
# Declare the AppDbContext
@scoped
class AppDbContext(AsyncSession, DataContext):
def __init__(self):
super().__init__(AsyncSQLEngineContext.get_client(), expire_on_commit=False)
# Create two models
class Hero(SQLModel):
id: int | None = None
name: str
class City(SQLModel):
id: int | None = None
city_name: str
# Create Repositories for each Model
@scoped
class HeroRepository(object):
def __init__(self, context: AppDbContext):
self.context = context
async def save(self, hero: Hero):
self.context.add(hero)
async def get_by_id(self, hero_id: int) -> Hero | None:
return await self.context.get(Hero, hero_id)
@scoped
class CityRepository(object):
def __init__(self, context: AppDbContext):
self.context = context
async def save(self, city: City):
self.context.add(city)
async def get_by_id(self, city_id: int) -> City | None:
return await self.context.get(City, city_id)
@scoped
class UnitOfWork(object):
def __init__(self, context: AppDbContext):
self.context = context
async def start(self):
await self.context.begin()
async def commit(self):
await self.context.commit()
async def rollback(self):
await self.context.rollback()
# alternatively run this block as a transaction
@asynccontextmanager
async def transaction(self):
try:
await self.start()
yield
except Exception as e:
await self.rollback()
raise e
finally:
await self.commit()
# This can be part of our service layer, for example
@scoped
class ChangeHeroNameService:
def __init__(self, repository: HeroRepository, uow: UnitOfWork):
self.repository = repository
self.uow = uow
async def change_hero_name(self, hero_id: int):
async with self.uow.transaction:
hero = await self.repository.get_by_id(hero_id)
if hero is not None:
hero.name = "New Hero Name"
# As we run in a transaction, the changes to hero
# will be committed at block end
# Now wire up in a controller
@controller
class HeroController:
hero_service: ChangeHeroNameService
@post("/{hero_id}")
async def change_hero_name(self, hero_id: int = Path()):
await self.hero_service.change_hero_name(hero_id)
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
Built Distribution
Hashes for wintrymodels-1.0.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | b3e01ec9a1cd12bc66b86a378cdb5ff907c16060b12d5ab2f516c6da5f3a941b |
|
MD5 | f289b0993615a42b2b561ba28b6d7321 |
|
BLAKE2b-256 | 2af2fdfb1e321eb60b31a34f556a09810125936c7c3b049c366117a532996d42 |