dienakdz
87f2845483
- Cleaned up whitespace and formatting in various files including http.py, language.py, logger.py, safe_exec.py, and SQL migration scripts. - Consolidated import statements and removed unnecessary blank lines. - Updated logging configuration for better clarity. - Enhanced the safe execution code with improved error handling and logging. - Removed commented-out code and unnecessary variables in backfill_zero_trades.py and other scripts. - Added a pyproject.toml for Ruff and Vulture configuration. - Introduced requirements-dev.txt for development dependencies. - Removed commented-out stock entries in init.sql for cleaner migration scripts.
73 lines
1.6 KiB
Python
73 lines
1.6 KiB
Python
"""
|
|
Database Connection Utility - PostgreSQL Only
|
|
|
|
Provides unified interface for PostgreSQL database operations.
|
|
|
|
Usage:
|
|
from app.utils.db import get_db_connection
|
|
|
|
with get_db_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
|
|
row = cursor.fetchone()
|
|
conn.commit()
|
|
|
|
Configuration:
|
|
DATABASE_URL=postgresql://user:password@host:port/dbname
|
|
"""
|
|
|
|
# Re-export from PostgreSQL module
|
|
from app.utils.db_postgres import (
|
|
close_pool as close_db,
|
|
)
|
|
from app.utils.db_postgres import (
|
|
get_pg_connection as get_db_connection,
|
|
)
|
|
from app.utils.db_postgres import (
|
|
get_pg_connection_sync as get_db_connection_sync,
|
|
)
|
|
from app.utils.db_postgres import (
|
|
is_postgres_available,
|
|
)
|
|
|
|
|
|
def get_db_type() -> str:
|
|
"""Get database type (always postgresql)"""
|
|
return "postgresql"
|
|
|
|
|
|
def is_postgres() -> bool:
|
|
"""Check if using PostgreSQL (always True)"""
|
|
return True
|
|
|
|
|
|
def init_database():
|
|
"""
|
|
Initialize database connection.
|
|
Schema is created via migrations/init.sql on PostgreSQL container start.
|
|
"""
|
|
if is_postgres_available():
|
|
from app.utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
logger.info("PostgreSQL connection verified")
|
|
else:
|
|
raise RuntimeError("Cannot connect to PostgreSQL. Check DATABASE_URL.")
|
|
|
|
|
|
# Legacy alias
|
|
def close_db_connection():
|
|
"""Legacy alias for close_db"""
|
|
pass
|
|
|
|
|
|
__all__ = [
|
|
"get_db_connection",
|
|
"get_db_connection_sync",
|
|
"close_db_connection",
|
|
"init_database",
|
|
"close_db",
|
|
"get_db_type",
|
|
"is_postgres",
|
|
]
|