# SQLRepositoryGenerator
## Description
SQLRepositoryGenerator is a wrapper above [SQLAlchemy](https://www.sqlalchemy.org/) to allow the generation of
Repository class from
SQLAlchemy models.
This way one can concentrate the repository code to non standard SQL query and have the common one auto generated.
## Installation
```bash
pip install sql-repository-generator
```
## Usage
two scenario are considered regarding the usage of the library:
- The most common one as base class for your repository to inherit
- A functional approach though function that generate repository class.
### Hierarchical
to use SQLRepositoryGenerator, it is needed to make a child class of one of the Repository base class
#### AsyncRepository
AsyncRepository is a base class that just hide the internal of query making for a given model
##### Example
```python
from typing import Annotated
from uuid import UUID, uuid4
from sqlalchemy import ForeignKey
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped, relationship
from sqlgen import AsyncRepository
UUID_PK = Annotated[UUID, mapped_column(primary_key=True)]
PROJECT_FK = Annotated[UUID, mapped_column(ForeignKey("project.id"))]
class Base(DeclarativeBase):
id: Mapped[UUID_PK] = mapped_column(default=uuid4)
class Host(Base):
__tablename__ = "host"
name: Mapped[str]
project_id: Mapped[PROJECT_FK]
project: Mapped["Project"] = relationship(back_populates="hosts")
class HostRepository(AsyncRepository):
cls = Host # Model to query
async def main(session: AsyncSession):
repository = HostRepository(session)
host = await repository.create(name="toto")
hosts = await repository.get_all()
```
#### AsyncConstrainedRepository (Deprecated)
AsyncConstrainedRepository handle more complex case where some constraint are needed to be fulfilled for every
interaction with the database.
Those constraints are defined on the child class using `bound_model` and `bound_models` class variable.
if they are set every query will filter using the link between `cls` and each `bound_model` to ensure they fulfill the
constraint.
In other word, considering we have a Database with a Customer, Project, Host and Request table.
and the following class
```python
class HostRepository(AsyncConstrainedRepository):
cls = Host
bound_model = Customer
```
every query will be automatically added a filter `Host.project.customer_id==customer_id`, this is also true for objects
creation however the logic is a bit different, instead of adding a filter to the select query, when creating an object,
every constraint will be validated though a dedicated query, raising an exception if failing, before adding the object.
##### Example
```python
from typing import Annotated
from uuid import UUID, uuid4
from sqlalchemy import ForeignKey
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped, relationship
from sqlgen import AsyncConstrainedRepository
UUID_PK = Annotated[UUID, mapped_column(primary_key=True)]
HOST_FK = Annotated[UUID, mapped_column(ForeignKey("host.id"))]
PROJECT_FK = Annotated[UUID, mapped_column(ForeignKey("project.id"))]
WEBSERVER_FK = Annotated[UUID, mapped_column(ForeignKey("webserver.id"))]
class Base(DeclarativeBase):
id: Mapped[UUID_PK] = mapped_column(default=uuid4)
class Request(Base):
webserver_id: Mapped[WEBSERVER_FK]
webserver: Mapped["Webserver"] = relationship(back_populates="requests")
class Webserver(Base):
__tablename__ = "webserver"
host_id: Mapped[HOST_FK]
host: Mapped["Host"] = relationship(back_populates="webservers")
requests: Mapped[list["Request"]] = relationship(back_populates="webserver", cascade="all, delete-orphan")
class Host(Base):
__tablename__ = "host"
name: Mapped[str]
project_id: Mapped[PROJECT_FK]
project: Mapped["Project"] = relationship(back_populates="hosts")
webservers: Mapped[list["Webserver"]] = relationship(back_populates="host", cascade="all, delete-orphan")
class Project(Base):
__tablename__ = "project"
hosts: Mapped[list["Host"]] = relationship(back_populates="project", cascade="all, delete-orphan")
class ProjectBoundRepository(AsyncConstrainedRepository):
bound_model = Project
class WebserverBoundRepository(AsyncConstrainedRepository):
bound_model = Webserver
class RequestRepository(ProjectBoundRepository, WebserverBoundRepository):
cls = Request
async def main(session: AsyncSession):
project_id = uuid4()
webserver_id = uuid4()
repository = RequestRepository(session, project_id=project_id, webserver_id=webserver_id)
request = await repository.create(name="toto") # check that webserver_id is bound to project_id
requests = await repository.get_all() # filtered by Request.webserver.host.project.id == project_id
```
#### AsyncObjectBoundRepository (Deprecated)
AsyncObjectBoundRepository allows to have a repository filtered for a specific object_id:
##### Example
```python
from typing import Annotated
from uuid import UUID, uuid4
from sqlalchemy import ForeignKey
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped, relationship
from sqlgen import AsyncObjectBoundRepository
UUID_PK = Annotated[UUID, mapped_column(primary_key=True)]
HOST_FK = Annotated[UUID, mapped_column(ForeignKey("host.id"))]
PROJECT_FK = Annotated[UUID, mapped_column(ForeignKey("project.id"))]
class Base(DeclarativeBase):
id: Mapped[UUID_PK] = mapped_column(default=uuid4)
class Webserver(Base):
__tablename__ = "webserver"
host_id: Mapped[HOST_FK]
host: Mapped["Host"] = relationship(back_populates="webservers")
class Host(Base):
__tablename__ = "host"
name: Mapped[str]
project_id: Mapped[PROJECT_FK]
project: Mapped["Project"] = relationship(back_populates="hosts")
webservers: Mapped[list["Webserver"]] = relationship(back_populates="host", cascade="all, delete-orphan")
class Project(Base):
__tablename__ = "project"
hosts: Mapped[list["Host"]] = relationship(back_populates="project", cascade="all, delete-orphan")
class WebserverRepository(AsyncObjectBoundRepository):
cls = Webserver # Model to query
bound_model = Project
async def main(session: AsyncSession):
project_id = uuid4()
repository = WebserverRepository(session, project_id)
host = await repository.create(name="toto") # Not Filtered
hosts = await repository.get_all() # filtered by Webserver.host.project.id == project_id
```
### Functional
three helper function are defined to generate repository classes:
- make_async_repository_class_for
- make_async_object_bound_repository_class_for
- make_async_constrained_repository_class_for
each of those function return an appropriate repository class for the parameters given to them.
Here is an example of their usage:
```python
# Async Repository
from sqlgen import make_async_repository_class_for, make_async_constrained_repository_class_for,
make_async_object_bound_repository_class_for
repository_class = make_async_repository_class_for(VulnerabilityClass)
repository = repository_class(session)
assert isinstance(repository, AsyncRepository)
# Object Bound Repository
repository_class = make_async_object_bound_repository_class_for(VulnerabilityClass, Project)
repository = repository_class(session, project_id)
assert isinstance(repository, AsyncObjectBoundRepository)
# Constrained Repository
repository_class = make_async_constrained_repository_class_for(VulnerabilityClass, Request, [Project])
repository = repository_class(session, request_id=request_id, project_id=project_id)
assert isinstance(repository, AsyncConstrainedRepository)
```
## Support
Any help is welcome. you can either:
- [create an issue](https://gitlab.com/Tagashy/sqlgen/issues/new)
- look for TODO in the code and provide a MR with changes
- provide a MR for support of new class
## Roadmap
- Make a public python package
## Authors and acknowledgment
Currently, solely developed by Tagashy but any help is welcomed and will be credited here.
## License
See the [LICENSE](LICENSE) file for licensing information as it pertains to
files in this repository.
Raw data
{
"_id": null,
"home_page": null,
"name": "sql-repository-generator",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": "sql, sqlalchemy, repository, generator",
"author": null,
"author_email": "Baboin Arthur <baboin.arthur@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/15/a4/3b284718278d756d1bc3d63500c34fe245eb03a8cc1a2af4e212b4fe43b7/sql_repository_generator-1.2.0.tar.gz",
"platform": null,
"description": "\n\n# SQLRepositoryGenerator\n\n\n\n\n\n## Description\n\nSQLRepositoryGenerator is a wrapper above [SQLAlchemy](https://www.sqlalchemy.org/) to allow the generation of\nRepository class from\nSQLAlchemy models.\n\nThis way one can concentrate the repository code to non standard SQL query and have the common one auto generated.\n\n## Installation\n\n```bash\npip install sql-repository-generator\n```\n\n## Usage\n\ntwo scenario are considered regarding the usage of the library:\n\n- The most common one as base class for your repository to inherit\n- A functional approach though function that generate repository class.\n\n### Hierarchical\n\nto use SQLRepositoryGenerator, it is needed to make a child class of one of the Repository base class\n\n#### AsyncRepository\n\nAsyncRepository is a base class that just hide the internal of query making for a given model\n\n##### Example\n\n```python\nfrom typing import Annotated\nfrom uuid import UUID, uuid4\n\nfrom sqlalchemy import ForeignKey\nfrom sqlalchemy.ext.asyncio import AsyncSession\nfrom sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped, relationship\n\nfrom sqlgen import AsyncRepository\n\nUUID_PK = Annotated[UUID, mapped_column(primary_key=True)]\nPROJECT_FK = Annotated[UUID, mapped_column(ForeignKey(\"project.id\"))]\n\n\nclass Base(DeclarativeBase):\n id: Mapped[UUID_PK] = mapped_column(default=uuid4)\n\n\nclass Host(Base):\n __tablename__ = \"host\"\n name: Mapped[str]\n project_id: Mapped[PROJECT_FK]\n project: Mapped[\"Project\"] = relationship(back_populates=\"hosts\")\n\n\nclass HostRepository(AsyncRepository):\n cls = Host # Model to query\n\n\nasync def main(session: AsyncSession):\n repository = HostRepository(session)\n host = await repository.create(name=\"toto\")\n hosts = await repository.get_all()\n\n```\n\n#### AsyncConstrainedRepository (Deprecated)\n\nAsyncConstrainedRepository handle more complex case where some constraint are needed to be fulfilled for every\ninteraction with the database.\nThose constraints are defined on the child class using `bound_model` and `bound_models` class variable.\nif they are set every query will filter using the link between `cls` and each `bound_model` to ensure they fulfill the\nconstraint.\n\nIn other word, considering we have a Database with a Customer, Project, Host and Request table.\nand the following class\n\n```python\nclass HostRepository(AsyncConstrainedRepository):\n cls = Host\n bound_model = Customer\n```\n\nevery query will be automatically added a filter `Host.project.customer_id==customer_id`, this is also true for objects\ncreation however the logic is a bit different, instead of adding a filter to the select query, when creating an object,\nevery constraint will be validated though a dedicated query, raising an exception if failing, before adding the object.\n\n##### Example\n\n```python\nfrom typing import Annotated\nfrom uuid import UUID, uuid4\n\nfrom sqlalchemy import ForeignKey\nfrom sqlalchemy.ext.asyncio import AsyncSession\nfrom sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped, relationship\n\nfrom sqlgen import AsyncConstrainedRepository\n\nUUID_PK = Annotated[UUID, mapped_column(primary_key=True)]\nHOST_FK = Annotated[UUID, mapped_column(ForeignKey(\"host.id\"))]\nPROJECT_FK = Annotated[UUID, mapped_column(ForeignKey(\"project.id\"))]\nWEBSERVER_FK = Annotated[UUID, mapped_column(ForeignKey(\"webserver.id\"))]\n\n\nclass Base(DeclarativeBase):\n id: Mapped[UUID_PK] = mapped_column(default=uuid4)\n\n\nclass Request(Base):\n webserver_id: Mapped[WEBSERVER_FK]\n webserver: Mapped[\"Webserver\"] = relationship(back_populates=\"requests\")\n\n\nclass Webserver(Base):\n __tablename__ = \"webserver\"\n\n host_id: Mapped[HOST_FK]\n host: Mapped[\"Host\"] = relationship(back_populates=\"webservers\")\n requests: Mapped[list[\"Request\"]] = relationship(back_populates=\"webserver\", cascade=\"all, delete-orphan\")\n\n\nclass Host(Base):\n __tablename__ = \"host\"\n name: Mapped[str]\n project_id: Mapped[PROJECT_FK]\n project: Mapped[\"Project\"] = relationship(back_populates=\"hosts\")\n webservers: Mapped[list[\"Webserver\"]] = relationship(back_populates=\"host\", cascade=\"all, delete-orphan\")\n\n\nclass Project(Base):\n __tablename__ = \"project\"\n hosts: Mapped[list[\"Host\"]] = relationship(back_populates=\"project\", cascade=\"all, delete-orphan\")\n\n\nclass ProjectBoundRepository(AsyncConstrainedRepository):\n bound_model = Project\n\n\nclass WebserverBoundRepository(AsyncConstrainedRepository):\n bound_model = Webserver\n\n\nclass RequestRepository(ProjectBoundRepository, WebserverBoundRepository):\n cls = Request\n\n\nasync def main(session: AsyncSession):\n project_id = uuid4()\n webserver_id = uuid4()\n repository = RequestRepository(session, project_id=project_id, webserver_id=webserver_id)\n request = await repository.create(name=\"toto\") # check that webserver_id is bound to project_id\n requests = await repository.get_all() # filtered by Request.webserver.host.project.id == project_id\n\n```\n\n#### AsyncObjectBoundRepository (Deprecated)\n\nAsyncObjectBoundRepository allows to have a repository filtered for a specific object_id:\n\n##### Example\n\n```python\nfrom typing import Annotated\nfrom uuid import UUID, uuid4\n\nfrom sqlalchemy import ForeignKey\nfrom sqlalchemy.ext.asyncio import AsyncSession\nfrom sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped, relationship\n\nfrom sqlgen import AsyncObjectBoundRepository\n\nUUID_PK = Annotated[UUID, mapped_column(primary_key=True)]\nHOST_FK = Annotated[UUID, mapped_column(ForeignKey(\"host.id\"))]\nPROJECT_FK = Annotated[UUID, mapped_column(ForeignKey(\"project.id\"))]\n\n\nclass Base(DeclarativeBase):\n id: Mapped[UUID_PK] = mapped_column(default=uuid4)\n\n\nclass Webserver(Base):\n __tablename__ = \"webserver\"\n\n host_id: Mapped[HOST_FK]\n host: Mapped[\"Host\"] = relationship(back_populates=\"webservers\")\n\n\nclass Host(Base):\n __tablename__ = \"host\"\n name: Mapped[str]\n project_id: Mapped[PROJECT_FK]\n project: Mapped[\"Project\"] = relationship(back_populates=\"hosts\")\n webservers: Mapped[list[\"Webserver\"]] = relationship(back_populates=\"host\", cascade=\"all, delete-orphan\")\n\n\nclass Project(Base):\n __tablename__ = \"project\"\n hosts: Mapped[list[\"Host\"]] = relationship(back_populates=\"project\", cascade=\"all, delete-orphan\")\n\n\nclass WebserverRepository(AsyncObjectBoundRepository):\n cls = Webserver # Model to query\n bound_model = Project\n\n\nasync def main(session: AsyncSession):\n project_id = uuid4()\n repository = WebserverRepository(session, project_id)\n host = await repository.create(name=\"toto\") # Not Filtered\n hosts = await repository.get_all() # filtered by Webserver.host.project.id == project_id\n\n```\n\n### Functional\n\nthree helper function are defined to generate repository classes:\n\n- make_async_repository_class_for\n- make_async_object_bound_repository_class_for\n- make_async_constrained_repository_class_for\n\neach of those function return an appropriate repository class for the parameters given to them.\nHere is an example of their usage:\n\n```python\n# Async Repository\nfrom sqlgen import make_async_repository_class_for, make_async_constrained_repository_class_for,\n\nmake_async_object_bound_repository_class_for\n\nrepository_class = make_async_repository_class_for(VulnerabilityClass)\nrepository = repository_class(session)\nassert isinstance(repository, AsyncRepository)\n# Object Bound Repository\nrepository_class = make_async_object_bound_repository_class_for(VulnerabilityClass, Project)\nrepository = repository_class(session, project_id)\nassert isinstance(repository, AsyncObjectBoundRepository)\n# Constrained Repository\nrepository_class = make_async_constrained_repository_class_for(VulnerabilityClass, Request, [Project])\nrepository = repository_class(session, request_id=request_id, project_id=project_id)\nassert isinstance(repository, AsyncConstrainedRepository)\n```\n\n## Support\n\nAny help is welcome. you can either:\n\n- [create an issue](https://gitlab.com/Tagashy/sqlgen/issues/new)\n- look for TODO in the code and provide a MR with changes\n- provide a MR for support of new class\n\n## Roadmap\n\n- Make a public python package\n\n## Authors and acknowledgment\n\nCurrently, solely developed by Tagashy but any help is welcomed and will be credited here.\n\n## License\n\nSee the [LICENSE](LICENSE) file for licensing information as it pertains to\nfiles in this repository.\n",
"bugtrack_url": null,
"license": null,
"summary": "Library to generate repository for SQLAlchemy based on ORM Models",
"version": "1.2.0",
"project_urls": {
"Bug Tracker": "https://gitlab.com/Tagashy/easysql/issues",
"Documentation": "https://tagashy.gitlab.io/sqlgen/",
"Homepage": "https://gitlab.com/Tagashy/sqlgen",
"Issues": "https://gitlab.com/Tagashy/sqlgen/issues",
"Repository": "https://gitlab.com/Tagashy/sqlgen.git"
},
"split_keywords": [
"sql",
" sqlalchemy",
" repository",
" generator"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "a5d625de886bda1100115c719311ac7c542108a06dcbfdaf096f9a893a15213c",
"md5": "846293f74d4a2f6596eeb312b3b49ea0",
"sha256": "d39479d43dff5822cae55b93ee27221daa3576d15d9d6b7a1043dbac642ae659"
},
"downloads": -1,
"filename": "sql_repository_generator-1.2.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "846293f74d4a2f6596eeb312b3b49ea0",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 27645,
"upload_time": "2024-12-11T14:57:09",
"upload_time_iso_8601": "2024-12-11T14:57:09.216836Z",
"url": "https://files.pythonhosted.org/packages/a5/d6/25de886bda1100115c719311ac7c542108a06dcbfdaf096f9a893a15213c/sql_repository_generator-1.2.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "15a43b284718278d756d1bc3d63500c34fe245eb03a8cc1a2af4e212b4fe43b7",
"md5": "36fe32d56ddb4b8c749aec99703eda78",
"sha256": "3312b2bbab8b85572e48e2e2010b677a96d7147e6123be0c831c7614cc89bb53"
},
"downloads": -1,
"filename": "sql_repository_generator-1.2.0.tar.gz",
"has_sig": false,
"md5_digest": "36fe32d56ddb4b8c749aec99703eda78",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 18353,
"upload_time": "2024-12-11T14:57:11",
"upload_time_iso_8601": "2024-12-11T14:57:11.603771Z",
"url": "https://files.pythonhosted.org/packages/15/a4/3b284718278d756d1bc3d63500c34fe245eb03a8cc1a2af4e212b4fe43b7/sql_repository_generator-1.2.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-11 14:57:11",
"github": false,
"gitlab": true,
"bitbucket": false,
"codeberg": false,
"gitlab_user": "Tagashy",
"gitlab_project": "easysql",
"lcname": "sql-repository-generator"
}
JSON