CODEWITHSUNDEEP
pythonpytestfastapi
lr_optim
lr_optim

Reputation: 381

FastAPI psycopg2 error when creating database for tests

I'm trying to test my database in my FastAPI app.

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.db import Base, get_db, get_session

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base.metadata.create_all(bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_get_users():
    response = client.get("/users")
    assert response.status_code == 200

This gives me an error sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) could not translate host name "db" to address: Name or service not known.

In my app.db, this is how I define the engine:

engine = create_engine(settings.DATABASE_URL, pool_pre_ping=True, echo=True, echo_pool=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def get_session():
    with Session(engine) as session:
        yield session

def get_db() -> Generator:
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()

I also tried using get_session instead of get_db, then the error is: no such table: users. Any ideas what is wrong with my setup?

Upvotes: 0

Views: 733

Answers (1)

Ashkan Goleh Pour
Ashkan Goleh Pour

Reputation: 522

if you have an already table in your database , you have to use MetaData to map them like:

from sqlalchemy import create_engine, MetaData
meta = MetaData(bind=engine)
meta.reflect(views=True)

table_name = meta.tables['table_name']

Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = Session()

def get_db():
   db = session
   try:
        yield db
   finally:
        db.close()

Upvotes: 1

Related Questions