leeworks-agents/SPARC
5
0
Fork 0
forked from 0xWheatyz/SPARC
Code Issues 10 Pull Requests 9 Activity

refactor(db): use shared pooled DatabaseClient singleton instead of per-call instances

Browse Source 
- Replace get_db_client() creating new DatabaseClient on every call with a
  module-level singleton initialized once at startup via init_db_client()
- Add init_db_client() and close_db_client() lifecycle functions called
  from FastAPI lifespan handler
- Migrate all DatabaseClient methods from legacy self.connect()/self.conn
  to pooled self.get_conn() context manager for thread-safe connection reuse
- Pool is properly torn down on application shutdown

Closes leeworks-agents/SPARC#7

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
agent-company
2026-03-26 04:15:03 +00:00
parent 35d105b14e
commit d366443b38
3 changed files with 186 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SPARC/api.py
+5 -1
View File
@@ -21,11 +21,13 @@ from SPARC.auth import (
TokenResponse, TokenResponse,
UserResponse, UserResponse,
check_jwt_secret, check_jwt_secret,
close_db_client,
create_tokens, create_tokens,
decode_token, decode_token,
get_current_admin, get_current_admin,
get_current_user, get_current_user,
get_db_client, get_db_client,
init_db_client,
) )
from SPARC.types import BatchAnalysisResult, CompanyAnalysisResult from SPARC.types import BatchAnalysisResult, CompanyAnalysisResult
@@ -155,6 +157,7 @@ async def lifespan(app: FastAPI):
"""Initialize resources on startup, clean up on shutdown.""" """Initialize resources on startup, clean up on shutdown."""
global _analyzer global _analyzer
check_jwt_secret() check_jwt_secret()
init_db_client()
_analyzer = CompanyAnalyzer() _analyzer = CompanyAnalyzer()
# Mark any jobs that were running/pending before the restart as failed # Mark any jobs that were running/pending before the restart as failed
from SPARC.database import DatabaseClient from SPARC.database import DatabaseClient
@@ -167,8 +170,9 @@ async def lifespan(app: FastAPI):
logging.getLogger(__name__).warning("Marked %d stale jobs as failed on startup", stale) logging.getLogger(__name__).warning("Marked %d stale jobs as failed on startup", stale)
_db.close() _db.close()
yield yield
# Cleanup if needed # Cleanup
_analyzer = None _analyzer = None
close_db_client()
app = FastAPI( app = FastAPI(
SPARC/auth.py
+29 -4
View File
@@ -146,11 +146,36 @@ def decode_token(token: str) -> Optional[TokenPayload]:
return None return None
# Shared database client singleton, initialized at startup via init_db_client()
_db_client: DatabaseClient | None = None
def init_db_client() -> None:
"""Initialize the shared database client. Call once at app startup."""
global _db_client
_db_client = DatabaseClient(config.database_url)
_db_client.connect()
def close_db_client() -> None:
"""Close the shared database client. Call at app shutdown."""
global _db_client
if _db_client:
_db_client.close()
_db_client = None
def get_db_client() -> DatabaseClient: def get_db_client() -> DatabaseClient:
"""Get database client for auth operations.""" """Get the shared pooled database client for auth operations.
client = DatabaseClient(config.database_url)
client.connect() Returns the module-level singleton DatabaseClient. If not yet initialized
return client (e.g., during tests), creates a new instance as a fallback.
"""
global _db_client
if _db_client is None:
_db_client = DatabaseClient(config.database_url)
_db_client.connect()
return _db_client
async def get_current_user( async def get_current_user(
SPARC/database.py
+152 -165
View File
@@ -221,8 +221,6 @@ class DatabaseClient:
Returns: Returns:
Cached message dict if found, None otherwise Cached message dict if found, None otherwise
""" """
self.connect()
prompt_hash = self.hash_prompt(prompt) prompt_hash = self.hash_prompt(prompt)
query = """ query = """
@@ -245,10 +243,11 @@ class DatabaseClient:
query += " ORDER BY timestamp DESC LIMIT 1" query += " ORDER BY timestamp DESC LIMIT 1"
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor: with self.get_conn() as conn:
cursor.execute(query, params) with conn.cursor(cursor_factory=RealDictCursor) as cursor:
result = cursor.fetchone() cursor.execute(query, params)
return dict(result) if result else None result = cursor.fetchone()
return dict(result) if result else None
def store_message( def store_message(
self, self,
@@ -276,33 +275,32 @@ class DatabaseClient:
Returns: Returns:
The ID of the inserted record The ID of the inserted record
""" """
self.connect()
prompt_hash = self.hash_prompt(prompt) prompt_hash = self.hash_prompt(prompt)
with self.conn.cursor() as cursor: with self.get_conn() as conn:
cursor.execute( with conn.cursor() as cursor:
""" cursor.execute(
INSERT INTO llm_messages """
(prompt, prompt_hash, response, company_name, analysis_type, model, metadata, token_usage, is_cached) INSERT INTO llm_messages
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) (prompt, prompt_hash, response, company_name, analysis_type, model, metadata, token_usage, is_cached)
RETURNING id VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
""", RETURNING id
( """,
prompt, (
prompt_hash, prompt,
response, prompt_hash,
company_name, response,
analysis_type, company_name,
model, analysis_type,
json.dumps(metadata) if metadata else None, model,
json.dumps(token_usage) if token_usage else None, json.dumps(metadata) if metadata else None,
is_cached, json.dumps(token_usage) if token_usage else None,
), is_cached,
) ),
)
message_id = cursor.fetchone()[0] message_id = cursor.fetchone()[0]
self.conn.commit() conn.commit()
return message_id return message_id
@@ -324,8 +322,6 @@ class DatabaseClient:
Returns: Returns:
List of message dictionaries List of message dictionaries
""" """
self.connect()
query = "SELECT * FROM llm_messages WHERE 1=1" query = "SELECT * FROM llm_messages WHERE 1=1"
params = [] params = []
@@ -340,9 +336,10 @@ class DatabaseClient:
query += " ORDER BY timestamp DESC LIMIT %s OFFSET %s" query += " ORDER BY timestamp DESC LIMIT %s OFFSET %s"
params.extend([limit, offset]) params.extend([limit, offset])
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor: with self.get_conn() as conn:
cursor.execute(query, params) with conn.cursor(cursor_factory=RealDictCursor) as cursor:
return [dict(row) for row in cursor.fetchall()] cursor.execute(query, params)
return [dict(row) for row in cursor.fetchall()]
def get_analytics(self, days: int = 30) -> Dict: def get_analytics(self, days: int = 30) -> Dict:
"""Get analytics on message usage. """Get analytics on message usage.
@@ -353,53 +350,52 @@ class DatabaseClient:
Returns: Returns:
Dictionary with analytics data Dictionary with analytics data
""" """
self.connect() with self.get_conn() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
# Total messages
cursor.execute(
"""
SELECT COUNT(*) as total_messages
FROM llm_messages
WHERE timestamp >= NOW() - INTERVAL '%s days'
""",
(days,),
)
total = cursor.fetchone()["total_messages"]
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor: # Messages by company
# Total messages cursor.execute(
cursor.execute( """
""" SELECT company_name, COUNT(*) as count
SELECT COUNT(*) as total_messages FROM llm_messages
FROM llm_messages WHERE timestamp >= NOW() - INTERVAL '%s days'
WHERE timestamp >= NOW() - INTERVAL '%s days' GROUP BY company_name
""", ORDER BY count DESC
(days,), LIMIT 10
) """,
total = cursor.fetchone()["total_messages"] (days,),
)
by_company = cursor.fetchall()
# Messages by company # Messages by type
cursor.execute( cursor.execute(
""" """
SELECT company_name, COUNT(*) as count SELECT analysis_type, COUNT(*) as count
FROM llm_messages FROM llm_messages
WHERE timestamp >= NOW() - INTERVAL '%s days' WHERE timestamp >= NOW() - INTERVAL '%s days'
GROUP BY company_name GROUP BY analysis_type
ORDER BY count DESC ORDER BY count DESC
LIMIT 10 """,
""", (days,),
(days,), )
) by_type = cursor.fetchall()
by_company = cursor.fetchall()
# Messages by type return {
cursor.execute( "total_messages": total,
""" "by_company": [dict(row) for row in by_company],
SELECT analysis_type, COUNT(*) as count "by_type": [dict(row) for row in by_type],
FROM llm_messages "period_days": days,
WHERE timestamp >= NOW() - INTERVAL '%s days' }
GROUP BY analysis_type
ORDER BY count DESC
""",
(days,),
)
by_type = cursor.fetchall()
return {
"total_messages": total,
"by_company": [dict(row) for row in by_company],
"by_type": [dict(row) for row in by_type],
"period_days": days,
}
# Patent Cache Methods # Patent Cache Methods
@@ -650,25 +646,23 @@ class DatabaseClient:
Returns: Returns:
Created user dict or None if email exists Created user dict or None if email exists
""" """
self.connect()
password_hash = self.hash_password(password) password_hash = self.hash_password(password)
try: try:
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor: with self.get_conn() as conn:
cursor.execute( with conn.cursor(cursor_factory=RealDictCursor) as cursor:
""" cursor.execute(
INSERT INTO users (email, password_hash, role) """
VALUES (%s, %s, %s) INSERT INTO users (email, password_hash, role)
RETURNING id, email, role, created_at VALUES (%s, %s, %s)
""", RETURNING id, email, role, created_at
(email, password_hash, role), """,
) (email, password_hash, role),
user = cursor.fetchone() )
self.conn.commit() user = cursor.fetchone()
conn.commit()
return dict(user) if user else None return dict(user) if user else None
except psycopg2.errors.UniqueViolation: except psycopg2.errors.UniqueViolation:
self.conn.rollback()
return None return None
def authenticate_user(self, email: str, password: str) -> Optional[Dict]: def authenticate_user(self, email: str, password: str) -> Optional[Dict]:
@@ -681,23 +675,22 @@ class DatabaseClient:
Returns: Returns:
User dict if authenticated, None otherwise User dict if authenticated, None otherwise
""" """
self.connect() with self.get_conn() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
cursor.execute(
"SELECT * FROM users WHERE email = %s",
(email,),
)
user = cursor.fetchone()
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor: if user and self.verify_password(password, user["password_hash"]):
cursor.execute( return {
"SELECT * FROM users WHERE email = %s", "id": user["id"],
(email,), "email": user["email"],
) "role": user["role"],
user = cursor.fetchone() "created_at": user["created_at"],
}
if user and self.verify_password(password, user["password_hash"]): return None
return {
"id": user["id"],
"email": user["email"],
"role": user["role"],
"created_at": user["created_at"],
}
return None
def get_user_by_id(self, user_id: int) -> Optional[Dict]: def get_user_by_id(self, user_id: int) -> Optional[Dict]:
"""Get a user by ID. """Get a user by ID.
@@ -708,15 +701,14 @@ class DatabaseClient:
Returns: Returns:
User dict or None User dict or None
""" """
self.connect() with self.get_conn() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor: cursor.execute(
cursor.execute( "SELECT id, email, role, created_at FROM users WHERE id = %s",
"SELECT id, email, role, created_at FROM users WHERE id = %s", (user_id,),
(user_id,), )
) user = cursor.fetchone()
user = cursor.fetchone() return dict(user) if user else None
return dict(user) if user else None
def get_user_by_email(self, email: str) -> Optional[Dict]: def get_user_by_email(self, email: str) -> Optional[Dict]:
"""Get a user by email. """Get a user by email.
@@ -727,15 +719,14 @@ class DatabaseClient:
Returns: Returns:
User dict or None User dict or None
""" """
self.connect() with self.get_conn() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor: cursor.execute(
cursor.execute( "SELECT id, email, role, created_at FROM users WHERE email = %s",
"SELECT id, email, role, created_at FROM users WHERE email = %s", (email,),
(email,), )
) user = cursor.fetchone()
user = cursor.fetchone() return dict(user) if user else None
return dict(user) if user else None
def get_all_users(self, limit: int = 100, offset: int = 0) -> List[Dict]: def get_all_users(self, limit: int = 100, offset: int = 0) -> List[Dict]:
"""Get all users (admin only). """Get all users (admin only).
@@ -747,19 +738,18 @@ class DatabaseClient:
Returns: Returns:
List of user dicts List of user dicts
""" """
self.connect() with self.get_conn() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor: cursor.execute(
cursor.execute( """
""" SELECT id, email, role, created_at
SELECT id, email, role, created_at FROM users
FROM users ORDER BY created_at DESC
ORDER BY created_at DESC LIMIT %s OFFSET %s
LIMIT %s OFFSET %s """,
""", (limit, offset),
(limit, offset), )
) return [dict(row) for row in cursor.fetchall()]
return [dict(row) for row in cursor.fetchall()]
def update_user_role(self, user_id: int, role: str) -> Optional[Dict]: def update_user_role(self, user_id: int, role: str) -> Optional[Dict]:
"""Update a user's role (admin only). """Update a user's role (admin only).
@@ -771,20 +761,19 @@ class DatabaseClient:
Returns: Returns:
Updated user dict or None Updated user dict or None
""" """
self.connect() with self.get_conn() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor: cursor.execute(
cursor.execute( """
""" UPDATE users
UPDATE users SET role = %s, updated_at = CURRENT_TIMESTAMP
SET role = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s
WHERE id = %s RETURNING id, email, role, created_at
RETURNING id, email, role, created_at """,
""", (role, user_id),
(role, user_id), )
) user = cursor.fetchone()
user = cursor.fetchone() conn.commit()
self.conn.commit()
return dict(user) if user else None return dict(user) if user else None
def delete_user(self, user_id: int) -> bool: def delete_user(self, user_id: int) -> bool:
@@ -796,12 +785,11 @@ class DatabaseClient:
Returns: Returns:
True if deleted True if deleted
""" """
self.connect() with self.get_conn() as conn:
with conn.cursor() as cursor:
with self.conn.cursor() as cursor: cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
cursor.execute("DELETE FROM users WHERE id = %s", (user_id,)) deleted = cursor.rowcount > 0
deleted = cursor.rowcount > 0 conn.commit()
self.conn.commit()
return deleted return deleted
def get_user_count(self) -> int: def get_user_count(self) -> int:
@@ -810,8 +798,7 @@ class DatabaseClient:
Returns: Returns:
Number of users Number of users
""" """
self.connect() with self.get_conn() as conn:
with conn.cursor() as cursor:
with self.conn.cursor() as cursor: cursor.execute("SELECT COUNT(*) FROM users")
cursor.execute("SELECT COUNT(*) FROM users") return cursor.fetchone()[0]
return cursor.fetchone()[0]