SQLArgon
Wrapper around SQLAlchemy async session, core and Postgres native features
Version: 0.3.2
Documentation: https://performancemedia.github.io/sqlargon/
Repository: https://github.com/performancemedia/sqlargon
About
This library provides glue code to use sqlalchemy async sessions, core queries and orm models from one object which provides somewhat of repository pattern. This solution has few advantages:
- no need to pass
sessionobject to every function/method. It is stored (and optionally injected) in repository object - write data access queries in one place
- no need to import
insert,update,delete,selectfrom sqlalchemy over and over again - Implicit cast of results to
.scalars().all()or.one() - Your view model (e.g. FastAPI routes) does not need to know about the underlying storage. Repository class can be replaced at any moment any object providing similar interface.
Usage
import sqlalchemy as sa
from sqlalchemy.orm import Mapped
from sqlargon import GUID, GenerateUUID, Database, Base, SQLAlchemyRepository
db = Database(url=...)
class User(Base):
id = sa.Column(
GUID(), primary_key=True, server_default=GenerateUUID(), nullable=False
)
name: Mapped[str] = sa.Column(sa.Unicode(255))
class UserRepository(SQLAlchemyRepository[User]):
async def get_user_by_name(self, name: str):
# custom user function
return await self.select().filter_by(name=name).one()
user_repository = UserRepository(db)
# select
await user_repository.all()
await user_repository.list(User.name == "test", User.id >= 18)
# insert
user = await user_repository.insert({"name": "test"}).one()
await user_repository.commit()
# delete
await user_repository.delete().filter(name="John").one()
# custom sqlalchemy core functions
users = await user_repository.select().join(...).filter(
User.name == "test"
).filter_by(...).order_by(User.created_at).limit(2).all()
Sessions
Manager object needs sqlalchemy.ext.asyncio.AsyncSession, but it's possible
to provide the session object by yourself, by subclassing Manager class e.g.
from sqlargon import Database, SQLAlchemyRepository
from fastapi import Depends
db = Database(url="sqlite+aiosqlite:///:memory:")
class UserRepository(SQLAlchemyRepository[User]):
...
from fastapi import FastAPI
app = FastAPI()
@app.get("/users")
async def get_users(user_repository: UserRepository = db.Depends(UserRepository)):
return await user_repository.all()
1 maintainer
Release files for sqlargon 0.3.2
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| sqlargon-0.3.2.tar.gz | 18.7 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| sqlargon-0.3.2-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 40.0 kB
Release files / sqlargon-0.3.2.tar.gz
| Download URL | sqlargon-0.3.2.tar.gz |
|---|---|
| Size | 18.7 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
c15b69fa26e971561b836cff871b0b7b63dcdb129289ac2fdb1efadf65371421
|
|
BLAKE2b-256 checksum How to use checksums |
454da947a565faa0963e253dceb0fce9b7d7e30dcb5ac078c548b73e235594d9
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/5.0.0 CPython/3.9.18
|
Release files / sqlargon-0.3.2-py3-none-any.whl
| Download URL | sqlargon-0.3.2-py3-none-any.whl |
|---|---|
| Size | 21.3 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
9d8ba2741dab2bf809532657768075c2be431ef2aa5b26deac017b33dbbd3c42
|
|
BLAKE2b-256 checksum How to use checksums |
adc218e4d58e5a4b9a001b5344e8c3702d13b39442102d571155f4506dce9d2f
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/5.0.0 CPython/3.9.18
|
