The fastapi-transactional from psymoney

fastapi-transactional's Introduction

목표: fastapi에서 사용할 수 있는 request 단위의 Transactional 애노테이션 만들기

가설

sqlalchemy의 async_scoped_session은 task 단위로 Session을 공유한다.
FastAPI는 request마다 task를 할당한다.
Service 레이어에서 async_scoped_session에 접근하면, 하위 repository에서 공유하는 동일한 Session을 제어할 수 있다.
애노테이션을 통해 로직을 반복되는 서비스 코드에 노출하지 않고 적용하기

fastapi-transactional's People

Contributors

Watchers

fastapi-transactional's Issues

테스트 목표

동시에 여러 request에 대해 task 단위 session 관리가 정상적으로 동작하는지 확인하기

목표 달성을 위한 테스트 코드 샘플 스니펫(httpx.AsyncClient 활용)

import pytest
import httpx
from fastapi import FastAPI
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

@pytest.mark.asyncio
async def test_multiple_requests():
    async with httpx.AsyncClient(app=app, base_url="http://test") as async_client:
        requests = [
            async_client.get(f"/items/{i}")
            for i in range(10)
        ]
        responses = await asyncio.gather(*requests)

    for response in responses:
        assert response.status_code == 200
        # assert 아이템 검증