server/database.py - master - Lexus_666/PIR_LAB_4

server/database.py
125 строк4 KB
Lexus-666
Lab_4_done added client + server + test run
25 дек 2025, 02:27
25 дек 2025, 02:271e2f432
import psycopg2
from psycopg2.extras import RealDictCursor
from common.config import config
import logging

logger = logging.getLogger(__name__)


class Database:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(Database, cls).__new__(cls)
            cls._instance._init_connection()
        return cls._instance

    def _init_connection(self):
        try:
            self.connection = psycopg2.connect(
                host=config.DB_HOST,
                port=config.DB_PORT,
                database=config.DB_NAME,
                user=config.DB_USER,
                password=config.DB_PASSWORD
            )
            self.connection.autocommit = True
            logger.info("Database connection established")
        except Exception as e:
            logger.error(f"Database connection failed: {e}")
            raise

    def execute_query(self, query: str, params: tuple = None):
        try:
            with self.connection.cursor(cursor_factory=RealDictCursor) as cursor:
                cursor.execute(query, params or ())
                if query.strip().upper().startswith('SELECT'):
                    return cursor.fetchall()
                elif query.strip().upper().startswith('INSERT'):
                    if 'RETURNING' in query.upper():
                        return cursor.fetchone()
                elif query.strip().upper().startswith('UPDATE'):
                    if 'RETURNING' in query.upper():
                        return cursor.fetchone()
                elif query.strip().upper().startswith('DELETE'):
                    if 'RETURNING' in query.upper():
                        return cursor.fetchone()
                return None
        except Exception as e:
            logger.error(f"Query execution failed: {e}")
            raise

    def is_request_processed(self, request_id: str) -> bool:
        query = "SELECT request_id FROM processed_requests WHERE request_id = %s"
        result = self.execute_query(query, (request_id,))
        return bool(result)

    def mark_request_processed(self, request_id: str):
        query = """
                INSERT INTO processed_requests (request_id)
                VALUES (%s) ON CONFLICT (request_id) DO NOTHING \
                """
        self.execute_query(query, (request_id,))

    def create_user(self, username: str, email: str, full_name: str = None):
        query = """
                INSERT INTO users (username, email, full_name)
                VALUES (%s, %s, %s) RETURNING id, username, email, full_name, created_at, updated_at \
                """
        return self.execute_query(query, (username, email, full_name))

    def get_user(self, user_id: int):
        query = "SELECT * FROM users WHERE id = %s"
        result = self.execute_query(query, (user_id,))
        if result:
            return result[0]
        return None

    def get_all_users(self):
        query = "SELECT * FROM users ORDER BY id"
        return self.execute_query(query)

    def update_user(self, user_id: int, username: str = None, email: str = None, full_name: str = None):

        updates = []
        params = []

        if username is not None:
            updates.append("username = %s")
            params.append(username)
        if email is not None:
            updates.append("email = %s")
            params.append(email)
        if full_name is not None:
            updates.append("full_name = %s")
            params.append(full_name)

        if not updates:
            return self.get_user(user_id)

        updates.append("updated_at = CURRENT_TIMESTAMP")
        params.append(user_id)

        set_clause = ", ".join(updates)
        query = f"""
        UPDATE users 
        SET {set_clause}
        WHERE id = %s
        RETURNING id, username, email, full_name, created_at, updated_at
        """

        return self.execute_query(query, tuple(params))

    def delete_user(self, user_id: int):
        query = """
                DELETE \
                FROM users
                WHERE id = %s RETURNING id, username, email, full_name, created_at, updated_at \
                """
        return self.execute_query(query, (user_id,))

    def validate_api_key(self, api_key: str) -> bool:
        query = "SELECT key FROM api_keys WHERE key = %s AND is_active = TRUE"
        result = self.execute_query(query, (api_key,))
        return bool(result)