60 lines
2 KiB
Python
60 lines
2 KiB
Python
# app/database.py
|
||
"""
|
||
Модуль для работы с базой данных PostgreSQL
|
||
Соответствует требованиям ТЗ по разработке базы данных
|
||
"""
|
||
import os
|
||
import psycopg2
|
||
from psycopg2.extras import RealDictCursor
|
||
from dotenv import load_dotenv
|
||
import time
|
||
|
||
load_dotenv()
|
||
|
||
class Database:
|
||
def __init__(self):
|
||
self.connection = None
|
||
self.max_retries = 3
|
||
self.retry_delay = 1
|
||
|
||
def get_connection(self):
|
||
"""Получение подключения к базе данных с повторными попытками"""
|
||
if self.connection is None or self.connection.closed:
|
||
for attempt in range(self.max_retries):
|
||
try:
|
||
self.connection = psycopg2.connect(
|
||
os.getenv('DATABASE_URL'),
|
||
cursor_factory=RealDictCursor
|
||
)
|
||
break
|
||
except psycopg2.OperationalError as e:
|
||
if attempt < self.max_retries - 1:
|
||
time.sleep(self.retry_delay)
|
||
continue
|
||
else:
|
||
raise e
|
||
return self.connection
|
||
|
||
def execute_query(self, query, params=None):
|
||
"""Выполнение SQL запроса с обработкой ошибок"""
|
||
conn = self.get_connection()
|
||
try:
|
||
with conn.cursor() as cursor:
|
||
cursor.execute(query, params)
|
||
if query.strip().upper().startswith('SELECT'):
|
||
return cursor.fetchall()
|
||
conn.commit()
|
||
return cursor.rowcount
|
||
except psycopg2.InterfaceError:
|
||
self.connection = None
|
||
raise
|
||
except Exception as e:
|
||
conn.rollback()
|
||
raise e
|
||
|
||
def close(self):
|
||
"""Закрытие соединения с базой данных"""
|
||
if self.connection and not self.connection.closed:
|
||
self.connection.close()
|
||
|
||
db = Database()
|