# Transactional-SQLAlchemy
## 개요
### 지원하는 트랜잭션 전파 방식
참조: [Transaction Propagation of Spring framework](https://docs.spring.io/spring-framework/reference/data-access/transaction/declarative/tx-propagation.html)
- `REQUIRED` : 이미 트랜잭션이 열린경우 기존의 세션을 사용하거나, 새로운 트랜잭션을 생성
- `REQUIRES_NEW` : 기존 트랜잭션을 무시하고 새롭게 생성
- `NESTED` : 기존 트랜잭션의 자식 트랜잭션을 생성
## 기능
트랜잭션 전파 방식 관리
Auto commit or Rollback (트랜잭션 사용 시)
auto session
동기/비동기 함수 모두 지원
## 사용법
### 1. transactional + auto session
1. 패키지 설치
- ver. sync
```bash
pip install transactional-sqlalchemy
```
- ver. async
```bash
pip install transactional-sqlalchemy[async]
```
2. 세션 핸들러 초기화
```python
from transactional_sqlalchemy import init_manager
from sqlalchemy.ext.asyncio import async_scoped_session
async_scoped_session_ = async_scoped_session(
async_session_factory, scopefunc=asyncio.current_task
)
init_manager(async_scoped_session_)
```
3. ITransactionalRepository를 상속하는 클래스 작성
- repository 레이어의 클래스 작성시, ITransactionalRepository를 상속
- `session`이라는 이름의 변수가 있는경우 만들어 두었던 세션을 할당
```python
from transactional_sqlalchemy import ITransactionalRepository, transactional
class PostRepository(ITransactionalRepository):
@transactional # or @transactional(propagation=Propagation.REQUIRES)
async def requires(self, post: Post, session: AsyncSession) -> None:
session.add(post)
...
@transactional(propagation=Propagation.REQUIRES_NEW)
async def requires_new(self, post: Post, session: AsyncSession) -> None: ...
@transactional(propagation=Propagation.NESTED)
async def nested(self, post: Post, session: AsyncSession) -> None: ...
async def auto_session_allocate(self, session:AsyncSession) -> None: ...
```
### 2. auto session without transactional
```python
from transactional_sqlalchemy import ISessionRepository
class PostRepository(ISessionRepository):
async def create(self, post: Post, *, session: AsyncSession = None) -> None: ...
```
### 3. 기본 CRUD Repository 사용
패키지에서 제공하는 기본 CRUD Repository 클래스를 상속하여 빠르게 Repository를 구현할 수 있습니다.
#### BaseCRUDRepository
기본적인 CRUD 연산을 제공하는 베이스 클래스입니다.
```python
from transactional_sqlalchemy import BaseCRUDRepository
from sqlalchemy.ext.asyncio import AsyncSession
from your_models import User
class UserRepository(BaseCRUDRepository[User]):
pass # 기본 CRUD 메서드들이 자동으로 사용 가능
# 사용 예시
user_repo = UserRepository()
# 기본 제공 메서드들
user = await user_repo.find_by_id(1, session=session)
users = await user_repo.find_all(session=session)
saved_user = await user_repo.save(new_user, session=session)
exists = await user_repo.exists_by_id(1, session=session)
count = await user_repo.count(session=session)
```
**제공되는 메서드:**
- `find_by_id(id, *, session)`: ID로 단일 모델 조회
- `find(where=None, *, session)`: 조건으로 단일 모델 조회
- `find_all(*, pageable=None, where=None, session)`: 전체 모델 조회 (페이징 지원)
- `find_all_by_id(ids, *, session)`: 여러 ID로 모델들 조회
- `save(model, *, session)`: 모델 저장/업데이트 (upsert)
- `exists(where=None, *, session)`: 모델 존재 여부 확인
- `exists_by_id(id, *, where=None, session)`: ID로 존재 여부 확인
- `count(*, where=None, session)`: 모델 개수 조회
#### BaseCRUDTransactionRepository
`BaseCRUDRepository`에 자동 트랜잭션 관리 기능이 추가된 클래스입니다.
```python
from transactional_sqlalchemy import BaseCRUDTransactionRepository, Propagation
from your_models import User
class UserTransactionRepository(BaseCRUDTransactionRepository[User]):
# 모든 메서드에 자동으로 @transactional 데코레이터가 적용됩니다
pass
# 사용 예시
user_repo = UserTransactionRepository()
# 트랜잭션이 자동으로 관리됩니다
user = await user_repo.find_by_id(1) # session 매개변수 불필요
saved_user = await user_repo.save(new_user) # 자동 커밋/롤백
```
#### 조건부 조회와 페이징
```python
from sqlalchemy import and_
from transactional_sqlalchemy import Pageable
# where 조건 사용
active_users = await user_repo.find_all(
where=and_(User.is_active == True, User.age >= 18),
session=session
)
# 페이징 사용
pageable = Pageable(offset=0, limit=10)
users_page = await user_repo.find_all(
pageable=pageable,
session=session
)
# 조건부 개수 조회
adult_count = await user_repo.count(
where=User.age >= 18,
session=session
)
```
#### 커스텀 메서드 추가
```python
class UserRepository(BaseCRUDTransactionRepository[User]):
@transactional(propagation=Propagation.REQUIRES)
async def find_by_email(self, email: str, *, session: AsyncSession) -> User | None:
return await self.find(where=User.email == email, session=session)
@transactional(propagation=Propagation.REQUIRES)
async def create_user_with_profile(self, user_data: dict, profile_data: dict, *, session: AsyncSession) -> User:
# 복잡한 비즈니스 로직
user = User(**user_data)
saved_user = await self.save(user, session=session)
profile = UserProfile(user_id=saved_user.id, **profile_data)
session.add(profile)
return saved_user
```
#### 고급 사용 예시
```python
from sqlalchemy import or_, desc
from transactional_sqlalchemy import BaseCRUDTransactionRepository, Propagation
class UserService(BaseCRUDTransactionRepository[User]):
@transactional(propagation=Propagation.REQUIRES)
async def search_users(self, keyword: str, *, session: AsyncSession) -> list[User]:
"""이름 또는 이메일로 사용자 검색"""
return await self.find_all(
where=or_(
User.name.ilike(f"%{keyword}%"),
User.email.ilike(f"%{keyword}%")
),
session=session
)
@transactional(propagation=Propagation.REQUIRES)
async def get_user_stats(self, *, session: AsyncSession) -> dict:
"""사용자 통계 조회"""
total_users = await self.count(session=session)
active_users = await self.count(where=User.is_active == True, session=session)
return {
"total": total_users,
"active": active_users,
"inactive": total_users - active_users
}
@transactional(propagation=Propagation.REQUIRES_NEW)
async def deactivate_user(self, user_id: int, *, session: AsyncSession) -> User:
"""사용자 비활성화 (새로운 트랜잭션)"""
user = await self.find_by_id(user_id, session=session)
if not user:
raise ValueError("User not found")
user.is_active = False
return await self.save(user, session=session)
```
Raw data
{
"_id": null,
"home_page": null,
"name": "transactional-sqlalchemy",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "sqlalchemy, transaction, async, database, orm",
"author": null,
"author_email": "alban <decade_vesper.8i@icloud.com>",
"download_url": "https://files.pythonhosted.org/packages/d4/dc/7dbdcbb303388d761cb07736264bf8617dba9bc749a9be155e547c93f131/transactional_sqlalchemy-0.1.9.tar.gz",
"platform": null,
"description": "# Transactional-SQLAlchemy\n\n## \uac1c\uc694\n\n### \uc9c0\uc6d0\ud558\ub294 \ud2b8\ub79c\uc7ad\uc158 \uc804\ud30c \ubc29\uc2dd\n\n\ucc38\uc870: [Transaction Propagation of Spring framework](https://docs.spring.io/spring-framework/reference/data-access/transaction/declarative/tx-propagation.html)\n\n- `REQUIRED` : \uc774\ubbf8 \ud2b8\ub79c\uc7ad\uc158\uc774 \uc5f4\ub9b0\uacbd\uc6b0 \uae30\uc874\uc758 \uc138\uc158\uc744 \uc0ac\uc6a9\ud558\uac70\ub098, \uc0c8\ub85c\uc6b4 \ud2b8\ub79c\uc7ad\uc158\uc744 \uc0dd\uc131\n- `REQUIRES_NEW` : \uae30\uc874 \ud2b8\ub79c\uc7ad\uc158\uc744 \ubb34\uc2dc\ud558\uace0 \uc0c8\ub86d\uac8c \uc0dd\uc131\n- `NESTED` : \uae30\uc874 \ud2b8\ub79c\uc7ad\uc158\uc758 \uc790\uc2dd \ud2b8\ub79c\uc7ad\uc158\uc744 \uc0dd\uc131\n\n## \uae30\ub2a5\n\n\ud2b8\ub79c\uc7ad\uc158 \uc804\ud30c \ubc29\uc2dd \uad00\ub9ac\n\nAuto commit or Rollback (\ud2b8\ub79c\uc7ad\uc158 \uc0ac\uc6a9 \uc2dc)\n\nauto session\n\n\ub3d9\uae30/\ube44\ub3d9\uae30 \ud568\uc218 \ubaa8\ub450 \uc9c0\uc6d0\n\n## \uc0ac\uc6a9\ubc95\n\n### 1. transactional + auto session\n\n1. \ud328\ud0a4\uc9c0 \uc124\uce58\n\n- ver. sync\n\n```bash\npip install transactional-sqlalchemy\n```\n\n- ver. async\n\n```bash\npip install transactional-sqlalchemy[async]\n```\n\n2. \uc138\uc158 \ud578\ub4e4\ub7ec \ucd08\uae30\ud654\n\n```python\nfrom transactional_sqlalchemy import init_manager\nfrom sqlalchemy.ext.asyncio import async_scoped_session\n\nasync_scoped_session_ = async_scoped_session(\n async_session_factory, scopefunc=asyncio.current_task\n)\n\ninit_manager(async_scoped_session_)\n\n```\n\n3. ITransactionalRepository\ub97c \uc0c1\uc18d\ud558\ub294 \ud074\ub798\uc2a4 \uc791\uc131\n\n- repository \ub808\uc774\uc5b4\uc758 \ud074\ub798\uc2a4 \uc791\uc131\uc2dc, ITransactionalRepository\ub97c \uc0c1\uc18d\n- `session`\uc774\ub77c\ub294 \uc774\ub984\uc758 \ubcc0\uc218\uac00 \uc788\ub294\uacbd\uc6b0 \ub9cc\ub4e4\uc5b4 \ub450\uc5c8\ub358 \uc138\uc158\uc744 \ud560\ub2f9\n\n```python\nfrom transactional_sqlalchemy import ITransactionalRepository, transactional\n\nclass PostRepository(ITransactionalRepository):\n @transactional # or @transactional(propagation=Propagation.REQUIRES)\n async def requires(self, post: Post, session: AsyncSession) -> None:\n session.add(post)\n ...\n\n @transactional(propagation=Propagation.REQUIRES_NEW)\n async def requires_new(self, post: Post, session: AsyncSession) -> None: ...\n\n @transactional(propagation=Propagation.NESTED)\n async def nested(self, post: Post, session: AsyncSession) -> None: ...\n\n async def auto_session_allocate(self, session:AsyncSession) -> None: ...\n```\n\n### 2. auto session without transactional\n\n```python\nfrom transactional_sqlalchemy import ISessionRepository\n\n\nclass PostRepository(ISessionRepository):\n\n async def create(self, post: Post, *, session: AsyncSession = None) -> None: ...\n```\n\n### 3. \uae30\ubcf8 CRUD Repository \uc0ac\uc6a9\n\n\ud328\ud0a4\uc9c0\uc5d0\uc11c \uc81c\uacf5\ud558\ub294 \uae30\ubcf8 CRUD Repository \ud074\ub798\uc2a4\ub97c \uc0c1\uc18d\ud558\uc5ec \ube60\ub974\uac8c Repository\ub97c \uad6c\ud604\ud560 \uc218 \uc788\uc2b5\ub2c8\ub2e4.\n\n#### BaseCRUDRepository\n\n\uae30\ubcf8\uc801\uc778 CRUD \uc5f0\uc0b0\uc744 \uc81c\uacf5\ud558\ub294 \ubca0\uc774\uc2a4 \ud074\ub798\uc2a4\uc785\ub2c8\ub2e4.\n\n```python\nfrom transactional_sqlalchemy import BaseCRUDRepository\nfrom sqlalchemy.ext.asyncio import AsyncSession\nfrom your_models import User\n\nclass UserRepository(BaseCRUDRepository[User]):\n pass # \uae30\ubcf8 CRUD \uba54\uc11c\ub4dc\ub4e4\uc774 \uc790\ub3d9\uc73c\ub85c \uc0ac\uc6a9 \uac00\ub2a5\n\n# \uc0ac\uc6a9 \uc608\uc2dc\nuser_repo = UserRepository()\n\n# \uae30\ubcf8 \uc81c\uacf5 \uba54\uc11c\ub4dc\ub4e4\nuser = await user_repo.find_by_id(1, session=session)\nusers = await user_repo.find_all(session=session)\nsaved_user = await user_repo.save(new_user, session=session)\nexists = await user_repo.exists_by_id(1, session=session)\ncount = await user_repo.count(session=session)\n```\n\n**\uc81c\uacf5\ub418\ub294 \uba54\uc11c\ub4dc:**\n\n- `find_by_id(id, *, session)`: ID\ub85c \ub2e8\uc77c \ubaa8\ub378 \uc870\ud68c\n- `find(where=None, *, session)`: \uc870\uac74\uc73c\ub85c \ub2e8\uc77c \ubaa8\ub378 \uc870\ud68c\n- `find_all(*, pageable=None, where=None, session)`: \uc804\uccb4 \ubaa8\ub378 \uc870\ud68c (\ud398\uc774\uc9d5 \uc9c0\uc6d0)\n- `find_all_by_id(ids, *, session)`: \uc5ec\ub7ec ID\ub85c \ubaa8\ub378\ub4e4 \uc870\ud68c\n- `save(model, *, session)`: \ubaa8\ub378 \uc800\uc7a5/\uc5c5\ub370\uc774\ud2b8 (upsert)\n- `exists(where=None, *, session)`: \ubaa8\ub378 \uc874\uc7ac \uc5ec\ubd80 \ud655\uc778\n- `exists_by_id(id, *, where=None, session)`: ID\ub85c \uc874\uc7ac \uc5ec\ubd80 \ud655\uc778\n- `count(*, where=None, session)`: \ubaa8\ub378 \uac1c\uc218 \uc870\ud68c\n\n#### BaseCRUDTransactionRepository\n\n`BaseCRUDRepository`\uc5d0 \uc790\ub3d9 \ud2b8\ub79c\uc7ad\uc158 \uad00\ub9ac \uae30\ub2a5\uc774 \ucd94\uac00\ub41c \ud074\ub798\uc2a4\uc785\ub2c8\ub2e4.\n\n```python\nfrom transactional_sqlalchemy import BaseCRUDTransactionRepository, Propagation\nfrom your_models import User\n\nclass UserTransactionRepository(BaseCRUDTransactionRepository[User]):\n # \ubaa8\ub4e0 \uba54\uc11c\ub4dc\uc5d0 \uc790\ub3d9\uc73c\ub85c @transactional \ub370\ucf54\ub808\uc774\ud130\uac00 \uc801\uc6a9\ub429\ub2c8\ub2e4\n pass\n\n# \uc0ac\uc6a9 \uc608\uc2dc\nuser_repo = UserTransactionRepository()\n\n# \ud2b8\ub79c\uc7ad\uc158\uc774 \uc790\ub3d9\uc73c\ub85c \uad00\ub9ac\ub429\ub2c8\ub2e4\nuser = await user_repo.find_by_id(1) # session \ub9e4\uac1c\ubcc0\uc218 \ubd88\ud544\uc694\nsaved_user = await user_repo.save(new_user) # \uc790\ub3d9 \ucee4\ubc0b/\ub864\ubc31\n```\n\n#### \uc870\uac74\ubd80 \uc870\ud68c\uc640 \ud398\uc774\uc9d5\n\n```python\nfrom sqlalchemy import and_\nfrom transactional_sqlalchemy import Pageable\n\n# where \uc870\uac74 \uc0ac\uc6a9\nactive_users = await user_repo.find_all(\n where=and_(User.is_active == True, User.age >= 18),\n session=session\n)\n\n# \ud398\uc774\uc9d5 \uc0ac\uc6a9\npageable = Pageable(offset=0, limit=10)\nusers_page = await user_repo.find_all(\n pageable=pageable,\n session=session\n)\n\n# \uc870\uac74\ubd80 \uac1c\uc218 \uc870\ud68c\nadult_count = await user_repo.count(\n where=User.age >= 18,\n session=session\n)\n```\n\n#### \ucee4\uc2a4\ud140 \uba54\uc11c\ub4dc \ucd94\uac00\n\n```python\nclass UserRepository(BaseCRUDTransactionRepository[User]):\n\n @transactional(propagation=Propagation.REQUIRES)\n async def find_by_email(self, email: str, *, session: AsyncSession) -> User | None:\n return await self.find(where=User.email == email, session=session)\n\n @transactional(propagation=Propagation.REQUIRES)\n async def create_user_with_profile(self, user_data: dict, profile_data: dict, *, session: AsyncSession) -> User:\n # \ubcf5\uc7a1\ud55c \ube44\uc988\ub2c8\uc2a4 \ub85c\uc9c1\n user = User(**user_data)\n saved_user = await self.save(user, session=session)\n\n profile = UserProfile(user_id=saved_user.id, **profile_data)\n session.add(profile)\n\n return saved_user\n```\n\n#### \uace0\uae09 \uc0ac\uc6a9 \uc608\uc2dc\n\n```python\nfrom sqlalchemy import or_, desc\nfrom transactional_sqlalchemy import BaseCRUDTransactionRepository, Propagation\n\nclass UserService(BaseCRUDTransactionRepository[User]):\n\n @transactional(propagation=Propagation.REQUIRES)\n async def search_users(self, keyword: str, *, session: AsyncSession) -> list[User]:\n \"\"\"\uc774\ub984 \ub610\ub294 \uc774\uba54\uc77c\ub85c \uc0ac\uc6a9\uc790 \uac80\uc0c9\"\"\"\n return await self.find_all(\n where=or_(\n User.name.ilike(f\"%{keyword}%\"),\n User.email.ilike(f\"%{keyword}%\")\n ),\n session=session\n )\n\n @transactional(propagation=Propagation.REQUIRES)\n async def get_user_stats(self, *, session: AsyncSession) -> dict:\n \"\"\"\uc0ac\uc6a9\uc790 \ud1b5\uacc4 \uc870\ud68c\"\"\"\n total_users = await self.count(session=session)\n active_users = await self.count(where=User.is_active == True, session=session)\n\n return {\n \"total\": total_users,\n \"active\": active_users,\n \"inactive\": total_users - active_users\n }\n\n @transactional(propagation=Propagation.REQUIRES_NEW)\n async def deactivate_user(self, user_id: int, *, session: AsyncSession) -> User:\n \"\"\"\uc0ac\uc6a9\uc790 \ube44\ud65c\uc131\ud654 (\uc0c8\ub85c\uc6b4 \ud2b8\ub79c\uc7ad\uc158)\"\"\"\n user = await self.find_by_id(user_id, session=session)\n if not user:\n raise ValueError(\"User not found\")\n\n user.is_active = False\n return await self.save(user, session=session)\n```\n",
"bugtrack_url": null,
"license": null,
"summary": "transactional management using sqlalchemy",
"version": "0.1.9",
"project_urls": {
"Bug Reports": "https://github.com/AlBaneo93/transactional_sqlalchemy/issues",
"Homepage": "https://github.com/AlBaneo93/transactional_sqlalchemy"
},
"split_keywords": [
"sqlalchemy",
" transaction",
" async",
" database",
" orm"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "50a110a7c6e7d6a9240654023433e6f5bc918f4d690d9cac6a8a9faf63583f31",
"md5": "9fafe74e6f01ceae65652f81c3e3b234",
"sha256": "c42b2b0b3c34326a2c7e48dd39eca4506c63c1c2e99d5630f6ccab97b181a545"
},
"downloads": -1,
"filename": "transactional_sqlalchemy-0.1.9-py3-none-any.whl",
"has_sig": false,
"md5_digest": "9fafe74e6f01ceae65652f81c3e3b234",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 26800,
"upload_time": "2025-08-12T05:12:26",
"upload_time_iso_8601": "2025-08-12T05:12:26.871509Z",
"url": "https://files.pythonhosted.org/packages/50/a1/10a7c6e7d6a9240654023433e6f5bc918f4d690d9cac6a8a9faf63583f31/transactional_sqlalchemy-0.1.9-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "d4dc7dbdcbb303388d761cb07736264bf8617dba9bc749a9be155e547c93f131",
"md5": "5134bfda0f9f4d422ef5a5e78d396034",
"sha256": "7d1abb8b916a1e7b57b3aa0ba5673a455cf2bc9b167c58b73f89be017c03fc6b"
},
"downloads": -1,
"filename": "transactional_sqlalchemy-0.1.9.tar.gz",
"has_sig": false,
"md5_digest": "5134bfda0f9f4d422ef5a5e78d396034",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 33243,
"upload_time": "2025-08-12T05:12:28",
"upload_time_iso_8601": "2025-08-12T05:12:28.299239Z",
"url": "https://files.pythonhosted.org/packages/d4/dc/7dbdcbb303388d761cb07736264bf8617dba9bc749a9be155e547c93f131/transactional_sqlalchemy-0.1.9.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-12 05:12:28",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "AlBaneo93",
"github_project": "transactional_sqlalchemy",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "transactional-sqlalchemy"
}
JSON
