db_manager.py - master - Askoline/Homework_DB2

db_manager.py
287 строк11 KB
Askoline
Initial commit: система управления клиентами
02 авг 2026, 14:28
02 авг 2026, 14:281333054
"""
Модуль для управления клиентами в PostgreSQL
"""
import psycopg2
from psycopg2 import sql


class ClientManager:
    """Класс для управления клиентами"""
    
    def __init__(self, conn):
        """Инициализация с подключением к БД"""
        self.conn = conn
    
    def create_db(self):
        """
        Функция, создающая структуру БД (таблицы).
        """
        with self.conn.cursor() as cur:
            # Создаем таблицу клиентов
            cur.execute("""
                CREATE TABLE IF NOT EXISTS clients (
                    id SERIAL PRIMARY KEY,
                    first_name VARCHAR(50) NOT NULL,
                    last_name VARCHAR(50) NOT NULL,
                    email VARCHAR(100) UNIQUE NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
            
            # Создаем таблицу телефонов
            cur.execute("""
                CREATE TABLE IF NOT EXISTS phones (
                    id SERIAL PRIMARY KEY,
                    client_id INTEGER NOT NULL REFERENCES clients(id) ON DELETE CASCADE,
                    phone_number VARCHAR(20) NOT NULL,
                    UNIQUE(client_id, phone_number)
                )
            """)
            
            self.conn.commit()
            print("✓ База данных успешно создана")
    
    def add_client(self, first_name, last_name, email, phones=None):
        """
        Функция, позволяющая добавить нового клиента.
        """
        with self.conn.cursor() as cur:
            try:
                # Добавляем клиента
                cur.execute("""
                    INSERT INTO clients (first_name, last_name, email)
                    VALUES (%s, %s, %s)
                    RETURNING id
                """, (first_name, last_name, email))
                
                client_id = cur.fetchone()[0]
                
                # Добавляем телефоны, если они есть
                if phones:
                    for phone in phones:
                        self.add_phone(client_id, phone)
                
                self.conn.commit()
                print(f"✓ Клиент '{first_name} {last_name}' добавлен с ID: {client_id}")
                return client_id
                
            except psycopg2.IntegrityError as e:
                self.conn.rollback()
                if "clients_email_key" in str(e):
                    raise ValueError(f"Клиент с email '{email}' уже существует")
                raise e
    
    def add_phone(self, client_id, phone):
        """
        Функция, позволяющая добавить телефон для существующего клиента.
        """
        with self.conn.cursor() as cur:
            # Проверяем существование клиента
            cur.execute("SELECT id FROM clients WHERE id = %s", (client_id,))
            if not cur.fetchone():
                raise ValueError(f"Клиент с ID {client_id} не найден")
            
            try:
                cur.execute("""
                    INSERT INTO phones (client_id, phone_number)
                    VALUES (%s, %s)
                """, (client_id, phone))
                self.conn.commit()
                print(f"✓ Телефон '{phone}' добавлен клиенту с ID: {client_id}")
                
            except psycopg2.IntegrityError:
                self.conn.rollback()
                raise ValueError(f"Телефон '{phone}' уже существует у этого клиента")
    
    def change_client(self, client_id, first_name=None, last_name=None, email=None):
        """
        Функция, позволяющая изменить данные о клиенте.
        """
        with self.conn.cursor() as cur:
            # Проверяем существование клиента
            cur.execute("SELECT * FROM clients WHERE id = %s", (client_id,))
            if not cur.fetchone():
                raise ValueError(f"Клиент с ID {client_id} не найден")
            
            # Формируем запрос на обновление
            updates = []
            params = []
            
            if first_name is not None:
                updates.append("first_name = %s")
                params.append(first_name)
            
            if last_name is not None:
                updates.append("last_name = %s")
                params.append(last_name)
            
            if email is not None:
                updates.append("email = %s")
                params.append(email)
            
            if not updates:
                print("ℹ Нет данных для обновления")
                return
            
            params.append(client_id)
            query = sql.SQL("UPDATE clients SET {} WHERE id = %s").format(
                sql.SQL(", ").join(map(sql.SQL, updates))
            )
            
            try:
                cur.execute(query, params)
                self.conn.commit()
                print(f"✓ Данные клиента с ID {client_id} обновлены")
                
            except psycopg2.IntegrityError:
                self.conn.rollback()
                raise ValueError(f"Клиент с email '{email}' уже существует")
    
    def delete_phone(self, client_id, phone):
        """
        Функция, позволяющая удалить телефон для существующего клиента.
        """
        with self.conn.cursor() as cur:
            # Проверяем существование клиента
            cur.execute("SELECT id FROM clients WHERE id = %s", (client_id,))
            if not cur.fetchone():
                raise ValueError(f"Клиент с ID {client_id} не найден")
            
            # Удаляем телефон
            cur.execute("""
                DELETE FROM phones 
                WHERE client_id = %s AND phone_number = %s
            """, (client_id, phone))
            
            if cur.rowcount == 0:
                raise ValueError(f"Телефон '{phone}' не найден у клиента")
            
            self.conn.commit()
            print(f"✓ Телефон '{phone}' удален у клиента с ID: {client_id}")
    
    def delete_client(self, client_id):
        """
        Функция, позволяющая удалить существующего клиента.
        """
        with self.conn.cursor() as cur:
            # Проверяем существование клиента
            cur.execute("SELECT id FROM clients WHERE id = %s", (client_id,))
            if not cur.fetchone():
                raise ValueError(f"Клиент с ID {client_id} не найден")
            
            # Удаляем клиента
            cur.execute("DELETE FROM clients WHERE id = %s", (client_id,))
            self.conn.commit()
            print(f"✓ Клиент с ID {client_id} удален")
    
    def find_client(self, first_name=None, last_name=None, email=None, phone=None):
        """
        Функция, позволяющая найти клиента по его данным.
        """
        with self.conn.cursor() as cur:
            # Строим запрос в зависимости от переданных параметров
            conditions = []
            params = []
            
            if first_name:
                conditions.append("c.first_name ILIKE %s")
                params.append(f"%{first_name}%")
            
            if last_name:
                conditions.append("c.last_name ILIKE %s")
                params.append(f"%{last_name}%")
            
            if email:
                conditions.append("c.email ILIKE %s")
                params.append(f"%{email}%")
            
            if phone:
                conditions.append("p.phone_number ILIKE %s")
                params.append(f"%{phone}%")
            
            if not conditions:
                print("ℹ Укажите хотя бы один параметр для поиска")
                return []
            
            # Выполняем поиск
            query = """
                SELECT DISTINCT c.id, c.first_name, c.last_name, c.email, c.created_at
                FROM clients c
                LEFT JOIN phones p ON c.id = p.client_id
                WHERE {}
                ORDER BY c.last_name, c.first_name
            """.format(" OR ".join(conditions))
            
            cur.execute(query, params)
            results = cur.fetchall()
            
            # Формируем результат с телефонами
            clients = []
            for row in results:
                cur.execute("SELECT phone_number FROM phones WHERE client_id = %s", (row[0],))
                phones = [p[0] for p in cur.fetchall()]
                
                clients.append({
                    'id': row[0],
                    'first_name': row[1],
                    'last_name': row[2],
                    'email': row[3],
                    'phones': phones,
                    'created_at': row[4]
                })
            
            return clients
    
    def get_all_clients(self):
        """
        Дополнительная функция: получение всех клиентов.
        """
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT id, first_name, last_name, email, created_at
                FROM clients
                ORDER BY last_name, first_name
            """)
            
            clients = []
            for row in cur.fetchall():
                cur.execute("SELECT phone_number FROM phones WHERE client_id = %s", (row[0],))
                phones = [p[0] for p in cur.fetchall()]
                
                clients.append({
                    'id': row[0],
                    'first_name': row[1],
                    'last_name': row[2],
                    'email': row[3],
                    'phones': phones,
                    'created_at': row[4]
                })
            
            return clients
    
    def get_client_by_id(self, client_id):
        """
        Дополнительная функция: получение клиента по ID.
        """
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT id, first_name, last_name, email, created_at
                FROM clients
                WHERE id = %s
            """, (client_id,))
            
            row = cur.fetchone()
            if not row:
                return None
            
            cur.execute("SELECT phone_number FROM phones WHERE client_id = %s", (client_id,))
            phones = [p[0] for p in cur.fetchall()]
            
            return {
                'id': row[0],
                'first_name': row[1],
                'last_name': row[2],
                'email': row[3],
                'phones': phones,
                'created_at': row[4]
            }