Spaces:
Sleeping
Sleeping
File size: 2,635 Bytes
b2be963 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from app.core.config import settings
# Check for Turso remote connection
is_turso = "turso.io" in settings.DATABASE_URL and settings.TURSO_AUTH_TOKEN
if is_turso:
# Extract hostname
hostname = settings.DATABASE_URL
if "://" in hostname:
hostname = hostname.split("://")[1]
if "?" in hostname:
hostname = hostname.split("?")[0]
from sqlalchemy.pool import NullPool
# Proxy to satisfy SQLAlchemy's requirement for create_function on SQLite connections
class LibsqlConnectionProxy:
def __init__(self, conn):
# Use __dict__ to avoid infinite recursion with __setattr__
self.__dict__["_conn"] = conn
def __getattr__(self, name):
return getattr(self._conn, name)
def __setattr__(self, name, value):
if name == "_conn":
self.__dict__["_conn"] = value
else:
setattr(self._conn, name, value)
def create_function(self, *args, **kwargs):
# libsql doesn't support create_function over HTTP, but SQLAlchemy expects it
return None
# Force HTTPS for stability on HF Spaces (avoids 505/WebSocket handshake errors)
def create_libsql_connection():
import libsql
url = f"https://{hostname}"
print(f">>> [DB] Establishing SECURE HTTP connection to Turso: {hostname}")
conn = libsql.connect(url, auth_token=settings.TURSO_AUTH_TOKEN)
return LibsqlConnectionProxy(conn)
# Use NullPool for Turso HTTPS to avoid "detached connection fairy" errors in multi-thread env
engine = create_engine(
"sqlite+libsql://",
creator=create_libsql_connection,
poolclass=NullPool,
echo=False,
)
else:
# Standard local SQLite
connect_args = {}
if "sqlite" in settings.DATABASE_URL or "libsql" in settings.DATABASE_URL:
connect_args["check_same_thread"] = False
engine = create_engine(
settings.DATABASE_URL,
pool_size=20,
max_overflow=max(10, 20), # Allow some overflow during peak
pool_timeout=30,
pool_recycle=1800,
connect_args=connect_args,
echo=False,
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
class Base(DeclarativeBase):
pass
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
|