test_sql_connection.py - develop - baelorsw/Alarma_Engine

test_sql_connection.py
81 строка3 KB
baelorsw
первый коннект с БД
11 май 2026, 20:58
11 май 2026, 20:588518776
# init_db.py
import pyodbc
import time

# Параметры подключения
DRIVER = '{ODBC Driver 17 for SQL Server}'
SERVER = 'localhost'
DATABASE = 'master'  # системная БД для создания новых
UID = 'sa'
PWD = 'Ss58545256'

# Подключение к master
def get_connection(database='master'):
    conn_str = f'DRIVER={DRIVER};SERVER={SERVER};DATABASE={database};UID={UID};PWD={PWD}'
    retry_count = 0
    while retry_count < 5:
        try:
            return pyodbc.connect(conn_str)
        except pyodbc.Error as e:
            print(f"Ошибка подключения к SQL Server: {e}")
            print("Ждём 5 секунд... Возможно, SQL Server ещё стартует.")
            time.sleep(5)
            retry_count += 1
    raise Exception("Не удалось подключиться к SQL Server после 5 попыток.")

# Создание базы
def create_database():
    conn = get_connection()
    conn.autocommit = True
    cursor = conn.cursor()

    cursor.execute("SELECT name FROM sys.databases WHERE name = 'SLA_Hackathon'")
    if cursor.fetchone():
        print("✅ База данных SLA_Hackathon уже существует.")
    else:
        cursor.execute("CREATE DATABASE SLA_Hackathon")
        print("🆕 База данных SLA_Hackathon создана.")

    conn.close()

# Создание таблицы
def create_tables():
    conn = get_connection('SLA_Hackathon')
    conn.autocommit = False
    cursor = conn.cursor()

    create_table_query = """
    IF NOT EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[dbo].[Requests]') AND type in (N'U'))
    BEGIN
        CREATE TABLE Requests (
            Id INT IDENTITY(1,1) PRIMARY KEY,
            Text NVARCHAR(MAX) NOT NULL,
            Category NVARCHAR(50) NOT NULL,
            Urgency NVARCHAR(50) NOT NULL,
            CreatedAt DATETIME2 DEFAULT GETUTCDATE(),
            SlaDeadline DATETIME2 NOT NULL,
            Status NVARCHAR(20) DEFAULT 'open',
            ClosedAt DATETIME2 NULL
        )

        CREATE INDEX IX_Requests_SlaDeadline ON Requests(SlaDeadline);
        CREATE INDEX IX_Requests_Status ON Requests(Status);
    END
    """
    try:
        cursor.execute(create_table_query)
        conn.commit()
        print("✅ Таблица Requests готова к работе.")
    except Exception as e:
        conn.rollback()
        print(f"❌ Ошибка при создании таблицы: {e}")
        raise
    finally:
        conn.close()

# Запуск инициализации
if __name__ == "__main__":
    print("🚀 Инициализация базы данных...")
    create_database()
    create_tables()
    print("🎉 Инициализация завершена. Готов к работе!")