app/utils/create_db.py - master - Kostyable/callmaster

app/utils/create_db.py
128 строк5 KB
Kostyable
feat: implement mvc model
29 окт 2024, 17:52
29 окт 2024, 17:520ad77c5
import psycopg2

from app.models.main import SpecializationEnum, OrderStatusEnum

def connect_to_db():
    try:
        conn = psycopg2.connect(
            dbname="microservices_db",
            user="postgres",
            password="qw27rty",
            host="localhost",
            port="5432"
        )
        conn.autocommit = True
        return conn
    except Exception as e:
        print(f"Ошибка подключения к базе данных: {e}")
        return None

def create_tables(conn):
    with conn.cursor() as cursor:
        try:
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS clients (
                    id SERIAL PRIMARY KEY,
                    last_name VARCHAR(100) NOT NULL,
                    first_name VARCHAR(100) NOT NULL,
                    email VARCHAR(255) NOT NULL UNIQUE,
                    phone_number VARCHAR(15) NOT NULL,
                    password VARCHAR(255) NOT NULL,
                    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                );
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS specializations (
                    id SERIAL PRIMARY KEY,
                    specialization_name VARCHAR(50) NOT NULL UNIQUE
                );
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS specialists (
                    id SERIAL PRIMARY KEY,
                    last_name VARCHAR(100) NOT NULL,
                    first_name VARCHAR(100) NOT NULL,
                    email VARCHAR(255) NOT NULL UNIQUE,
                    phone_number VARCHAR(15) NOT NULL,
                    password VARCHAR(255) NOT NULL,
                    specialization_id INT NOT NULL,
                    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (specialization_id) REFERENCES specializations(id)
                    ON DELETE CASCADE ON UPDATE CASCADE
                );
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS statuses (
                    id SERIAL PRIMARY KEY,
                    status_name VARCHAR(50) NOT NULL UNIQUE
                );
            """)
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS orders (
                    id SERIAL PRIMARY KEY,
                    client_pk INT NOT NULL,
                    specialist_pk INT NOT NULL,
                    specialization_id INT NOT NULL,
                    status_id INT NOT NULL,
                    comment TEXT,
                    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (client_pk) REFERENCES clients(id)
                    ON DELETE CASCADE ON UPDATE CASCADE,
                    FOREIGN KEY (specialist_pk) REFERENCES specialists(id)
                    ON DELETE CASCADE ON UPDATE CASCADE,
                    FOREIGN KEY (specialization_id) REFERENCES specializations(id)
                    ON DELETE CASCADE ON UPDATE CASCADE,
                    FOREIGN KEY (status_id) REFERENCES statuses(id)
                    ON DELETE CASCADE ON UPDATE CASCADE
                );
            """)
            print("Таблицы успешно созданы.")
        except Exception as e:
            print(f"Ошибка при создании таблиц: {e}")

def insert_specializations(conn):
    with conn.cursor() as cursor:
        try:
            for specialization in SpecializationEnum:
                cursor.execute("""
                    INSERT INTO specializations (specialization_name)
                    VALUES (%s) ON CONFLICT (specialization_name) DO NOTHING;
                """, (specialization.value,))
            print("Специализации успешно добавлены.")
        except Exception as e:
            print(f"Ошибка при добавлении специализаций: {e}")

def insert_statuses(conn):
    with conn.cursor() as cursor:
        try:
            for status in OrderStatusEnum:
                cursor.execute("""
                    INSERT INTO statuses (status_name)
                    VALUES (%s) ON CONFLICT (status_name) DO NOTHING;
                """, (status.value,))
            print("Статусы успешно добавлены.")
        except Exception as e:
            print(f"Ошибка при добавлении статусов: {e}")

def drop_tables(conn):
    with conn.cursor() as cursor:
        try:
            cursor.execute("DROP TABLE IF EXISTS orders, specialists, statuses, specializations, clients;")
            print("Таблицы успешно удалены.")
        except Exception as e:
            print(f"Ошибка при удалении таблиц: {e}")

def initialize_database():
    conn = connect_to_db()
    if conn:
        drop_tables(conn)
        create_tables(conn)
        insert_specializations(conn)
        insert_statuses(conn)
        conn.close()
        print("Инициализация базы данных завершена.")
    else:
        print("Не удалось подключиться к базе данных.")

if __name__ == "__main__":
    initialize_database()