Estoy trabajando en un proyecto FastAPI asíncrono y quiero conectarme a la base de datos durante las pruebas. Viniendo de Django, mi instinto fue crear dispositivos pytest que se encarguen de crear/eliminar la base de datos de prueba. Sin embargo, no pude encontrar mucha documentación sobre cómo hacer esto. Las instrucciones más completas que pude encontrar estaban en este tutorial , pero no me funcionan porque son todas sincrónicas. Soy algo nuevo en el desarrollo asíncrono, por lo que tengo problemas para adaptar el código para que funcione asíncronamente. Esto es lo que tengo hasta ahora:
import pytest from sqlalchemy.ext.asyncio import create_async_engine, session from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from sqlalchemy_utils import database_exists, create_database from fastapi.testclient import TestClient from app.core.db import get_session from app.main import app Base = declarative_base() @pytest.fixture(scope="session") def db_engine(): default_db = ( "postgresql+asyncpg://postgres:postgres@postgres:5432/postgres" ) test_db = "postgresql+asyncpg://postgres:postgres@postgres:5432/test" engine = create_async_engine(default_db) if not database_exists(test_db): # <- Getting error on this line create_database(test_db) Base.metadata.create_all(bind=engine) yield engine @pytest.fixture(scope="function") def db(db_engine): connection = db_engine.connect() # begin a non-ORM transaction connection.begin() # bind an individual Session to the connection Session = sessionmaker(bind=connection) db = Session() # db = Session(db_engine) yield db db.rollback() connection.close() @pytest.fixture(scope="function") def client(db): app.dependency_overrides[get_session] = lambda: db PREFIX = "/api/v1/my-endpoint" with TestClient(PREFIX, app) as c: yield cY este es el error que me sale:
E sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/14/xd2s) /usr/local/lib/python3.9/site-packages/sqlalchemy/util/_concurrency_py3k.py:67: MissingGreenlet¿Alguna idea de lo que tengo que hacer para arreglarlo?
Intenta usar el motor de sincronización con una sesión asíncrona . Tratar de usar:
from sqlalchemy.ext.asyncio import AsyncSession Session = sessionmaker(bind= connection, class_=AsyncSession)https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html