A library using the "Repository" pattern for working with a database. The library is based on Sqlalchemy 2.0 (DeclarativeBase).
Additionally, work with DTO based on pydantic 2.0 schemas (BaseModel)
Main Components:
DatabaseRepository: This is the main class that provides methods for interacting with the database. It includes methods for creating, reading, updating, and deleting data.
The repository supports working with nested models. You just need to describe the model in relationship (sqlalchemy) and pass the instance to the methods of create, update. The library itself will resolve the parent model and child models of any nesting.
You can flexibly configure filtering to get models from the database. All filtering is collected in the function DatabaseRepository._get(). You can add global filters for the entire class (switchable parameter), as well as for a specific repository function.
DTORepository: This class provides methods for working with DTOs, including validation and conversion to and from database models.
This class is a wrapper over the main DataBaseRepository. In our work we often need to transform models into their representations (DTO), especially when working with FastAPI.
This class can do everything a DatabaseRepository can do. He just knows how to work with pydantic schemas.
Important: When working with nested models (relationships), the schema must also be nested, where the field name matches the relationship name in the model, and the field must also be an instance of the class BaseModel.
Example:
```python
class Model1(OrmBase): <- OrmBase(DeclarativeBase)
id: Mapped[int]
name: Mapped[str]
model2: Mapped["Model2"] = relationship("Model2", back_populates="model1")
class Model2(OrmBase):
id: Mapped[int]
name: Mapped[str]
model1_id: Mapped[int]
model1: Mapped["Model1"] = relationship("Model1", back_populates="model2")
class Schema1(BaseModel):
id: int
name: str
class Schema2(BaseModel):
id: int
name: str
model1: Schema1 <- important field name and instance BaseModel
```
ConfigORM: This class is a global configuration class. You can change the configuration at any time.
Limit for outputting records from the database if no local limit is set when working with the repository.
Global filters for models. These filters are applied to all queries for all tables. Don't worry, if the table doesn't have such a field, it will simply be discarded and won't give an error. This filter can be enabled or disabled at any time in a repository instance.
For example, the most common filter: {'is_active': True}.
# Example of work:
```python
from pydantic import BaseModel
from sqlalchemy import Sequence
from sqlalchemy.orm import joinedload
from ormrepo.db_settings import config_orm
from ormrepo.orm import DatabaseRepository, DTORepository
from sqlalchemy.ext.asyncio import AsyncSession, AsyncEngine, create_async_engine, async_sessionmaker
from tests.models import TModel1, TModel1Schema, RelationModel1, TModel1Rel1Schema, RelationModel1Schema
from tests.session import uri
config_orm.configure(limit=100, global_filters={'is_active': True})
engine: AsyncEngine = create_async_engine(uri, echo=True)
sessionmaker = async_sessionmaker(engine)
async def get_session() -> AsyncSession:
"""Session Manager"""
async with sessionmaker() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
```
## Example of working with a repository:
```python
async def get_models() -> Sequence[TModel1]:
async with get_session() as session:
repo = DatabaseRepository(TModel1,
session,
local_filters=[TModel1.name.like("%model%")],
use_global_filters=False)
res = await repo.get_many(filters=[TModel1.id == 1],
load=[joinedload(TModel1.relation_models)],
relation_filters={RelationModel1: [RelationModel1.id.in_([1, 2, 3])]},
offset=10,
limit=10)
return res
```
## Example of working with a DTO:
```python
async def get_dto() -> list[BaseModel]:
async with get_session() as session:
repo = DTORepository(DatabaseRepository(TModel1, session),
TModel1Schema)
res = await repo.get_many(filters=[TModel1.id > 1])
return res
```
## Example of creation
```python
async def create_dto() -> TModel1Rel1Schema:
async with get_session() as session:
repo = DTORepository(DatabaseRepository(TModel1, session),
TModel1Rel1Schema)
return await repo.create(TModel1Rel1Schema(name='test',
serial=1,
relation_models=[
RelationModel1Schema(),
RelationModel1Schema(),
RelationModel1Schema()
]))
```
## Example of update
```python
async def update_dto() -> TModel1Rel1Schema:
async with get_session() as session:
repo = DTORepository(DatabaseRepository(TModel1, session),
TModel1Rel1Schema)
return await repo.update(1, TModel1Rel1Schema(name='test_new',
relation_models=None))
"""
Important:
There will be two changes to the databases here.
1: Model TModel1 name changes. Since in the update function, only those fields that are explicitly set are updated.
Because model_dump(exclude_unset=True).
2: Since we explicitly set relation_models, all records from the related table related to this record will be deleted!!!
"""
```
Raw data
{
"_id": null,
"home_page": "https://github.com/wahek/ormrepo",
"name": "ormrepo",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "sqlalchemy, pydantic, repository, orm, async, database, dto, pattern",
"author": "Ivan Maurin",
"author_email": "wahek1999@mail.ru",
"download_url": "https://files.pythonhosted.org/packages/93/42/ac92d4d738468a509e696dfc745f9367c681b267ffd6f7757cf5d9260df3/ormrepo-0.4.2.tar.gz",
"platform": null,
"description": "A library using the \"Repository\" pattern for working with a database. The library is based on Sqlalchemy 2.0 (DeclarativeBase).\nAdditionally, work with DTO based on pydantic 2.0 schemas (BaseModel)\n\nMain Components:\n\nDatabaseRepository: This is the main class that provides methods for interacting with the database. It includes methods for creating, reading, updating, and deleting data.\nThe repository supports working with nested models. You just need to describe the model in relationship (sqlalchemy) and pass the instance to the methods of create, update. The library itself will resolve the parent model and child models of any nesting.\nYou can flexibly configure filtering to get models from the database. All filtering is collected in the function DatabaseRepository._get(). You can add global filters for the entire class (switchable parameter), as well as for a specific repository function.\n\n\n\nDTORepository: This class provides methods for working with DTOs, including validation and conversion to and from database models.\nThis class is a wrapper over the main DataBaseRepository. In our work we often need to transform models into their representations (DTO), especially when working with FastAPI.\nThis class can do everything a DatabaseRepository can do. He just knows how to work with pydantic schemas.\nImportant: When working with nested models (relationships), the schema must also be nested, where the field name matches the relationship name in the model, and the field must also be an instance of the class BaseModel.\nExample:\n```python\nclass Model1(OrmBase): <- OrmBase(DeclarativeBase)\n id: Mapped[int]\n name: Mapped[str]\n model2: Mapped[\"Model2\"] = relationship(\"Model2\", back_populates=\"model1\")\n\nclass Model2(OrmBase):\n id: Mapped[int]\n name: Mapped[str]\n model1_id: Mapped[int]\n model1: Mapped[\"Model1\"] = relationship(\"Model1\", back_populates=\"model2\")\n \nclass Schema1(BaseModel):\n id: int\n name: str\nclass Schema2(BaseModel):\n id: int\n name: str\n model1: Schema1 <- important field name and instance BaseModel\n```\nConfigORM: This class is a global configuration class. You can change the configuration at any time.\nLimit for outputting records from the database if no local limit is set when working with the repository.\nGlobal filters for models. These filters are applied to all queries for all tables. Don't worry, if the table doesn't have such a field, it will simply be discarded and won't give an error. This filter can be enabled or disabled at any time in a repository instance.\nFor example, the most common filter: {'is_active': True}.\n\n# Example of work:\n```python\nfrom pydantic import BaseModel\nfrom sqlalchemy import Sequence\nfrom sqlalchemy.orm import joinedload\n\nfrom ormrepo.db_settings import config_orm\nfrom ormrepo.orm import DatabaseRepository, DTORepository\nfrom sqlalchemy.ext.asyncio import AsyncSession, AsyncEngine, create_async_engine, async_sessionmaker\n\nfrom tests.models import TModel1, TModel1Schema, RelationModel1, TModel1Rel1Schema, RelationModel1Schema\nfrom tests.session import uri\n\nconfig_orm.configure(limit=100, global_filters={'is_active': True})\n\nengine: AsyncEngine = create_async_engine(uri, echo=True)\nsessionmaker = async_sessionmaker(engine)\n\n\nasync def get_session() -> AsyncSession:\n \"\"\"Session Manager\"\"\"\n async with sessionmaker() as session:\n try:\n yield session\n await session.commit()\n except Exception:\n await session.rollback()\n raise\n finally:\n await session.close()\n```\n## Example of working with a repository:\n```python\nasync def get_models() -> Sequence[TModel1]:\n async with get_session() as session:\n repo = DatabaseRepository(TModel1,\n session,\n local_filters=[TModel1.name.like(\"%model%\")],\n use_global_filters=False)\n res = await repo.get_many(filters=[TModel1.id == 1],\n load=[joinedload(TModel1.relation_models)],\n relation_filters={RelationModel1: [RelationModel1.id.in_([1, 2, 3])]},\n offset=10,\n limit=10)\n return res\n```\n## Example of working with a DTO:\n```python\nasync def get_dto() -> list[BaseModel]:\n async with get_session() as session:\n repo = DTORepository(DatabaseRepository(TModel1, session),\n TModel1Schema)\n res = await repo.get_many(filters=[TModel1.id > 1])\n return res\n```\n## Example of creation\n```python\nasync def create_dto() -> TModel1Rel1Schema:\n async with get_session() as session:\n repo = DTORepository(DatabaseRepository(TModel1, session),\n TModel1Rel1Schema)\n return await repo.create(TModel1Rel1Schema(name='test',\n serial=1,\n relation_models=[\n RelationModel1Schema(),\n RelationModel1Schema(),\n RelationModel1Schema()\n ]))\n```\n## Example of update\n```python\nasync def update_dto() -> TModel1Rel1Schema:\n async with get_session() as session:\n repo = DTORepository(DatabaseRepository(TModel1, session),\n TModel1Rel1Schema)\n return await repo.update(1, TModel1Rel1Schema(name='test_new',\n relation_models=None))\n\"\"\"\nImportant:\nThere will be two changes to the databases here.\n1: Model TModel1 name changes. Since in the update function, only those fields that are explicitly set are updated.\nBecause model_dump(exclude_unset=True).\n2: Since we explicitly set relation_models, all records from the related table related to this record will be deleted!!!\n\"\"\"\n```",
"bugtrack_url": null,
"license": "MIT",
"summary": "Asynchronous repository pattern with SQLAlchemy 2.0 and Pydantic 2.0 support.",
"version": "0.4.2",
"project_urls": {
"Homepage": "https://github.com/wahek/ormrepo"
},
"split_keywords": [
"sqlalchemy",
" pydantic",
" repository",
" orm",
" async",
" database",
" dto",
" pattern"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "47c296ada7e7457d510c7831e787736224f44b01e1c7d216006f980c5ab6c80c",
"md5": "7ce0d88a119f600b866037cb366b6068",
"sha256": "be8da17a8965d4a7368125126d2dc9237ff53113cde057c155c31f577905cf79"
},
"downloads": -1,
"filename": "ormrepo-0.4.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7ce0d88a119f600b866037cb366b6068",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 12531,
"upload_time": "2025-07-24T12:40:11",
"upload_time_iso_8601": "2025-07-24T12:40:11.043888Z",
"url": "https://files.pythonhosted.org/packages/47/c2/96ada7e7457d510c7831e787736224f44b01e1c7d216006f980c5ab6c80c/ormrepo-0.4.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "9342ac92d4d738468a509e696dfc745f9367c681b267ffd6f7757cf5d9260df3",
"md5": "83cc75cf1a688257e014eaa1ce5a1c01",
"sha256": "e941375eda9850f6a644c1c30b288403dc785cef5ea1020722e69fdb65ba20a0"
},
"downloads": -1,
"filename": "ormrepo-0.4.2.tar.gz",
"has_sig": false,
"md5_digest": "83cc75cf1a688257e014eaa1ce5a1c01",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 11596,
"upload_time": "2025-07-24T12:40:12",
"upload_time_iso_8601": "2025-07-24T12:40:12.094567Z",
"url": "https://files.pythonhosted.org/packages/93/42/ac92d4d738468a509e696dfc745f9367c681b267ffd6f7757cf5d9260df3/ormrepo-0.4.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-24 12:40:12",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "wahek",
"github_project": "ormrepo",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "ormrepo"
}
JSON
