Coverage for app/db/supabaseDB.py: 80%

1from sqlalchemy import create_engine 

2from sqlalchemy.orm import declarative_base, sessionmaker 

3from sqlalchemy.pool import NullPool 

4from dotenv import load_dotenv 

5import os 

6 

7load_dotenv() 

8 

9USER = os.getenv("SUPABASE_DB_USER") 

10PASSWORD = os.getenv("SUPABASE_DB_PASSWORD") 

11HOST = os.getenv("SUPABASE_DB_HOST") 

12PORT = os.getenv("SUPABASE_DB_PORT") 

13DBNAME = os.getenv("SUPABASE_DB_NAME") 

14 

15# Construct the SQLAlchemy connection string 

16DATABASE_URL = f"postgresql+psycopg2://{USER}:{PASSWORD}@{HOST}:{PORT}/{DBNAME}?sslmode=require" 

17 

18# Create the SQLAlchemy engine 

19# engine = create_engine(DATABASE_URL) 

20# If using Transaction Pooler or Session Pooler, we want to ensure we disable SQLAlchemy client side pooling - 

21# https://docs.sqlalchemy.org/en/20/core/pooling.html#switching-pool-implementations 

22engine = create_engine(DATABASE_URL, poolclass=NullPool) 

23 

24SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False) 

25 

26Base = declarative_base() 

27 

28# Dependency to use in routes as middleware 

29def get_db(): 

30 db = SessionLocal() 

31 try: 

32 yield db 

33 finally: 

34 db.close() 

35 

36# Test the connection 

37# try: 

38# with engine.connect() as connection: 

39# print("Connection successful!") 

40# except Exception as e: 

41# print(f"Failed to connect: {e}")