Business
Jobs
  • About Us
  • Solutions
    • Job Postings
      Post your job and receive qualified candidates in 48h.
    • Candidate Assessments
      500+ technical and psychological tests, plus anti-fraud.
    • Headhunting
      Tailor-made executive search from start to finish.
    • Payroll + EOR
      Payroll dispersal and EOR across 15+ LATAM countries.
  • Pricing
  • Jobs

0

776
Views
¿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 answers
Answer question

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 Report
Answer question
Find remote jobs

Discover the new way to find a job!

Top jobs
Top job categories
Business
Post vacancy Pricing Sales
Legal
Terms and conditions Privacy policy
© 2026 PeakU Inc. All Rights Reserved.
Andres GPT
Show me some job opportunities
There's an error!