from sqlalchemy_service.base_db.db_configure import MySQLDBConfigurationfrom sqlalchemy.ext.asyncio import AsyncSessionfrom typing import AsyncGenerator
# Sqlalchemy-service
This is a library that simplifies working with database CRUD queries and connection management.
## Features
- A class to reduce the amount of code needed for database CRUD queries and connection management.
## Installation
- `pip install sqlalchemy-service[postresql]`
- `pip install sqlalchemy-service[mysql]`
- `pip install sqlalchemy-service[postgresql,fastapi]` for fastapi support(http exceptions and dependency injection)
## Usage
- Need environment set: POSTGRES_HOST, POSTGRES_DATABASE, POSTGRES_PASSWORD, POSTGRES_USER
```python3
import asyncio
from random import randint
from pydantic import BaseModel
from sqlalchemy import ScalarResult
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column as column
from sqlalchemy_service import Base
from sqlalchemy_service import BaseService
from sqlalchemy_service.base_db.base import ServiceEngine
from sqlalchemy_service.base_db.db_configure import MySQLDBConfiguration
class User(Base):
__tablename__ = "users"
id: Mapped[int] = column(primary_key=True)
name: Mapped[str]
class UserCreateSchema(BaseModel):
name: str
class UserUpdateSchema(BaseModel):
name: str | None = None
# You can manually set urls for engines, or use autodetect
engine_1 = ServiceEngine()
engine_2 = ServiceEngine(MySQLDBConfiguration().get_url())
class UserService[Table: User, int](BaseService):
base_table = User
engine = engine_1
# Redefine BaseService.get_session to connect with 2 engines
# Just don't rewrite it to connect with one engine
async def get_session(self):
if randint(0, 1):
return engine_1
return engine_2
async def create(self, schema: UserCreateSchema) -> User:
return await self._create(schema)
async def list(self, page=None, count=None) -> ScalarResult[User]:
return await self._get_list(page=page, count=count)
async def get(self, user_id: int) -> User:
"""Return user. If user not found, throws 404 HTTPException"""
return await self._get_one(id=user_id)
async def update(self, user_id: int, schema: UserUpdateSchema) -> User:
return await self._update(user_id, schema)
async def delete(self, user_id: int):
await self._delete(user_id)
async def count(self) -> int:
return await self._count()
async def count_with_name_like(self, name: str) -> int:
query = self._count_query()
query = self._query_like_filter(query, name=name)
return await self.session.scalar(query)
async def main():
# Use alembic to create database schema
async with UserService() as service:
print(await service.create(UserCreateSchema(name="test")))
asyncio.run(main())
```
# Updates
### 1.0.0:
- custom engine style
- more use-ready commands
Raw data
{
"_id": null,
"home_page": "https://github.com/Dianapp-online/sqlalchemy-service",
"name": "sqlalchemy-service",
"maintainer": "prostoLavr (Lawrence Naumov)",
"docs_url": null,
"requires_python": "<4.0,>=3.12",
"maintainer_email": "<prostolawr@gmail.com>",
"keywords": "sqlalchemy",
"author": "prostoLavr (Lawrence Naumov)",
"author_email": "<prostolawr@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/37/24/a313e097fc2fc3f147acab7e1bc774cc00f239ee3669eab48ad4dc4f12b6/sqlalchemy_service-1.0.0.tar.gz",
"platform": null,
"description": "from sqlalchemy_service.base_db.db_configure import MySQLDBConfigurationfrom sqlalchemy.ext.asyncio import AsyncSessionfrom typing import AsyncGenerator\n\n# Sqlalchemy-service\nThis is a library that simplifies working with database CRUD queries and connection management.\n\n## Features\n- A class to reduce the amount of code needed for database CRUD queries and connection management.\n\n## Installation\n- `pip install sqlalchemy-service[postresql]`\n- `pip install sqlalchemy-service[mysql]`\n- `pip install sqlalchemy-service[postgresql,fastapi]` for fastapi support(http exceptions and dependency injection)\n\n## Usage\n- Need environment set: POSTGRES_HOST, POSTGRES_DATABASE, POSTGRES_PASSWORD, POSTGRES_USER\n\n```python3\nimport asyncio\nfrom random import randint\n\nfrom pydantic import BaseModel\nfrom sqlalchemy import ScalarResult\nfrom sqlalchemy.orm import Mapped\nfrom sqlalchemy.orm import mapped_column as column\n\nfrom sqlalchemy_service import Base\nfrom sqlalchemy_service import BaseService\nfrom sqlalchemy_service.base_db.base import ServiceEngine\nfrom sqlalchemy_service.base_db.db_configure import MySQLDBConfiguration\n\n\nclass User(Base):\n __tablename__ = \"users\"\n id: Mapped[int] = column(primary_key=True)\n name: Mapped[str]\n\n\nclass UserCreateSchema(BaseModel):\n name: str\n\n\nclass UserUpdateSchema(BaseModel):\n name: str | None = None\n\n\n# You can manually set urls for engines, or use autodetect\nengine_1 = ServiceEngine()\nengine_2 = ServiceEngine(MySQLDBConfiguration().get_url())\n\n\nclass UserService[Table: User, int](BaseService):\n base_table = User\n engine = engine_1\n\n # Redefine BaseService.get_session to connect with 2 engines\n # Just don't rewrite it to connect with one engine\n async def get_session(self):\n if randint(0, 1):\n return engine_1\n return engine_2\n\n async def create(self, schema: UserCreateSchema) -> User:\n return await self._create(schema)\n\n async def list(self, page=None, count=None) -> ScalarResult[User]:\n return await self._get_list(page=page, count=count)\n\n async def get(self, user_id: int) -> User:\n \"\"\"Return user. If user not found, throws 404 HTTPException\"\"\"\n return await self._get_one(id=user_id)\n\n async def update(self, user_id: int, schema: UserUpdateSchema) -> User:\n return await self._update(user_id, schema)\n\n async def delete(self, user_id: int):\n await self._delete(user_id)\n\n async def count(self) -> int:\n return await self._count()\n\n async def count_with_name_like(self, name: str) -> int:\n query = self._count_query()\n query = self._query_like_filter(query, name=name)\n return await self.session.scalar(query)\n\n\nasync def main():\n # Use alembic to create database schema\n async with UserService() as service:\n print(await service.create(UserCreateSchema(name=\"test\")))\n\n\nasyncio.run(main())\n```\n\n# Updates\n### 1.0.0:\n- custom engine style\n- more use-ready commands",
"bugtrack_url": null,
"license": "LICENSE",
"summary": null,
"version": "1.0.0",
"project_urls": {
"Homepage": "https://github.com/Dianapp-online/sqlalchemy-service",
"Issues": "https://github.com/Dianapp-online/sqlalchemy-service/issues"
},
"split_keywords": [
"sqlalchemy"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "d503b2551041add5642ec8762d570425684f3c2e231134dd8e46df8e63a51a7d",
"md5": "2a6520cf7547c57c0bb51e2152cb4334",
"sha256": "a23f5596c303c98ac449df912fc456a3788f977d62a31f1022bd6c97bfa3c673"
},
"downloads": -1,
"filename": "sqlalchemy_service-1.0.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2a6520cf7547c57c0bb51e2152cb4334",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.12",
"size": 11678,
"upload_time": "2025-02-22T08:06:36",
"upload_time_iso_8601": "2025-02-22T08:06:36.060303Z",
"url": "https://files.pythonhosted.org/packages/d5/03/b2551041add5642ec8762d570425684f3c2e231134dd8e46df8e63a51a7d/sqlalchemy_service-1.0.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "3724a313e097fc2fc3f147acab7e1bc774cc00f239ee3669eab48ad4dc4f12b6",
"md5": "6d1b9abec2df4b67a48188e7f67ec5b5",
"sha256": "ab32905b090c9de962468a2a1f0713d22571998c0c4d385e62e103c9ed47cbbf"
},
"downloads": -1,
"filename": "sqlalchemy_service-1.0.0.tar.gz",
"has_sig": false,
"md5_digest": "6d1b9abec2df4b67a48188e7f67ec5b5",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.12",
"size": 9441,
"upload_time": "2025-02-22T08:06:38",
"upload_time_iso_8601": "2025-02-22T08:06:38.945558Z",
"url": "https://files.pythonhosted.org/packages/37/24/a313e097fc2fc3f147acab7e1bc774cc00f239ee3669eab48ad4dc4f12b6/sqlalchemy_service-1.0.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-02-22 08:06:38",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Dianapp-online",
"github_project": "sqlalchemy-service",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "sqlalchemy-service"
}
JSON
