Empresas
Empleos
  • Sobre nosotros
  • Soluciones
    • Publicación de vacantes
      Publica tu vacante y recibe candidatos calificados en 48h.
    • Evaluación de candidatos
      500+ pruebas técnicas y psicológicas, más anti-fraude.
    • Headhunting
      Búsqueda ejecutiva a la medida de principio a fin.
    • Nómina + EOR
      Dispersión de nómina y EOR en más de 15 países de LATAM.
  • Precios
  • Empleos

0

773
Vistas
¿Cómo usar la base de datos de prueba de PostgreSQL en pruebas asíncronas de FastAPI?

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 c

Y 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?

over 4 years ago · Santiago Trujillo
1 Respuestas
Responde la pregunta

0

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

over 4 years ago · Santiago Trujillo Denunciar
Responde la pregunta
Encuentra empleos remotos

¡Descubre la nueva forma de encontrar empleo!

Top de empleos
Top categorías de empleo
Empresas
Publicar vacante Precios Comercial
Legal
Términos y condiciones Política de privacidad
© 2026 PeakU Inc. All Rights Reserved.
Andres GPT
Recomiéndame algunas ofertas
Necesito ayuda