mirror of
https://github.com/grillazz/fastapi-sqlalchemy-asyncpg.git
synced 2026-01-17 11:40:39 +03:00
refactor: clean up imports and improve db_session fixture in test configuration
This commit is contained in:
@@ -106,7 +106,6 @@ class Settings(BaseSettings):
|
|||||||
path=self.POSTGRES_TEST_DB,
|
path=self.POSTGRES_TEST_DB,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@computed_field
|
@computed_field
|
||||||
@property
|
@property
|
||||||
def postgres_url(self) -> PostgresDsn:
|
def postgres_url(self) -> PostgresDsn:
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
from collections.abc import AsyncGenerator
|
from collections.abc import AsyncGenerator
|
||||||
from types import SimpleNamespace
|
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
@@ -8,7 +7,7 @@ from sqlalchemy import text
|
|||||||
from sqlalchemy.exc import ProgrammingError
|
from sqlalchemy.exc import ProgrammingError
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
from app.database import engine, test_engine, get_test_db, get_db
|
from app.database import engine, get_db, test_engine
|
||||||
from app.main import app
|
from app.main import app
|
||||||
from app.models.base import Base
|
from app.models.base import Base
|
||||||
from app.redis import get_redis
|
from app.redis import get_redis
|
||||||
@@ -23,6 +22,7 @@ from app.redis import get_redis
|
|||||||
def anyio_backend(request):
|
def anyio_backend(request):
|
||||||
return request.param
|
return request.param
|
||||||
|
|
||||||
|
|
||||||
def _create_db(conn) -> None:
|
def _create_db(conn) -> None:
|
||||||
"""Create the test database if it doesn't exist."""
|
"""Create the test database if it doesn't exist."""
|
||||||
try:
|
try:
|
||||||
@@ -66,7 +66,7 @@ async def start_db():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
async def db_session(start_db) -> AsyncGenerator[AsyncSession, Any]:
|
async def db_session() -> AsyncGenerator[AsyncSession, Any]:
|
||||||
"""
|
"""
|
||||||
Provide a transactional database session for each test function.
|
Provide a transactional database session for each test function.
|
||||||
Rolls back changes after the test.
|
Rolls back changes after the test.
|
||||||
|
|||||||
Reference in New Issue
Block a user