164 lines
7.0 KiB
Python
164 lines
7.0 KiB
Python
from data_base.data_base import async_session, Base, async_engine
|
||
import data_base.models.main_tables as mt
|
||
import json
|
||
import model_messege as mm
|
||
from sqlalchemy.future import select
|
||
from sqlalchemy import MetaData, inspect, text, insert, Table
|
||
import bcrypt
|
||
|
||
class JsonJober:
|
||
data = []
|
||
|
||
@classmethod
|
||
def create_table(cls, table_name):
|
||
""""Создание объектов таблицы"""
|
||
temp_table = []
|
||
if mt.Professions.__tablename__ == table_name:
|
||
for row in JsonJober.data[table_name]:
|
||
temp_table.append(mt.Professions(name=row))
|
||
print(f"Added to Professions: {row}")
|
||
elif mt.Positions.__tablename__ == table_name:
|
||
for row in JsonJober.data[table_name]:
|
||
temp_table.append(mt.Positions(name=row))
|
||
print(f"Added to Positions: {row}")
|
||
elif mt.Classification_professions.__tablename__ == table_name:
|
||
for row in JsonJober.data[table_name]:
|
||
temp_table.append(mt.Classification_professions(name=row))
|
||
print(f"Added to Classification_professions: {row}")
|
||
elif mt.Manufacturing_enterprises.__tablename__ == table_name:
|
||
for i, name in enumerate(JsonJober.data[table_name]["name"]):
|
||
inn = JsonJober.data[table_name]["inn"][i]
|
||
temp_table.append(mt.Manufacturing_enterprises(name=name, INN_company=inn))
|
||
print(f"Added to Manufacturing_enterprises: {name}, {inn}")
|
||
else:
|
||
print(f"Unknown table name: {table_name}")
|
||
return temp_table
|
||
|
||
@classmethod
|
||
def ReadJson(cls, path_file: str = ""):
|
||
"""Чтение json файла"""
|
||
JsonJober.data = []
|
||
mass_table = []
|
||
with open(path_file, "r", encoding="utf-8") as read_file:
|
||
JsonJober.data = json.load(read_file)
|
||
for table in JsonJober.data:
|
||
mass_table.append(JsonJober.create_table(table_name=table))
|
||
return mass_table
|
||
|
||
class Request():
|
||
@classmethod
|
||
async def load_preset(cls):
|
||
""""Загрузка пресета"""
|
||
await cls.craete_all_table()
|
||
mass_table = JsonJober.ReadJson(path_file="./data_base/preset/test.json")
|
||
await cls.insert_mass_table(mass_table=mass_table)
|
||
|
||
user_data = {
|
||
"inn": "123",
|
||
"password": bcrypt.hashpw("123".encode("utf-8"), bcrypt.gensalt()).decode('utf-8'),
|
||
"mail": "123",
|
||
"name_poo": "123"
|
||
}
|
||
await cls.insert_table(data={"data": [user_data]}, table_name="users")
|
||
|
||
|
||
@classmethod
|
||
async def insert_mass_table(cls, mass_table):
|
||
"""Добавление множества таблиц"""
|
||
async with async_session() as session:
|
||
async with session.begin():
|
||
for table in mass_table:
|
||
session.add_all(table)
|
||
await session.commit()
|
||
|
||
@classmethod
|
||
async def select_drop_down_list(cls):
|
||
"""Возвращает все выподающие списки"""
|
||
async with async_session() as session:
|
||
result = {}
|
||
tables = [mt.Professions, mt.Positions, mt.Classification_professions, mt.Manufacturing_enterprises]
|
||
for table in tables:
|
||
query = select(table)
|
||
result_proxy = await session.execute(query)
|
||
result[table.__tablename__] = [row._asdict() for row in result_proxy.fetchall()]
|
||
return result
|
||
|
||
@staticmethod
|
||
async def craete_all_table():
|
||
""""Создание всех таблиц"""
|
||
async with async_engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.drop_all)
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
|
||
@classmethod
|
||
async def login(cls, auth_request) -> mt.users:
|
||
"""Авторизация"""
|
||
async with async_session() as session:
|
||
stmt = select(mt.users).where(
|
||
mt.users.inn == auth_request.login,
|
||
)
|
||
result = await session.execute(stmt)
|
||
return result.scalars().first()
|
||
|
||
@classmethod
|
||
async def add_column(cls, table_name: str, column_name: str, data_type: str):
|
||
"""Добавление столбца в таблицу"""
|
||
async with async_engine.connect() as connection:
|
||
if data_type == 'String':
|
||
data_type_sql = "VARCHAR"
|
||
elif data_type == 'Integer':
|
||
data_type_sql = "INTEGER"
|
||
else:
|
||
raise ValueError("Unsupported data type")
|
||
|
||
alter_stmt = text(f'ALTER TABLE {table_name} ADD COLUMN {column_name} {data_type_sql}')
|
||
await connection.execute(alter_stmt)
|
||
await connection.commit()
|
||
|
||
@classmethod
|
||
async def drop_column(cls, table_name: str, column_name: str):
|
||
"""Удаление столбца из таблицы"""
|
||
async with async_engine.connect() as connection:
|
||
alter_stmt = text(f'ALTER TABLE {table_name} DROP COLUMN {column_name}')
|
||
await connection.execute(alter_stmt)
|
||
await connection.commit()
|
||
|
||
@classmethod
|
||
async def get_table_columns(cls, table_name: str):
|
||
"""Получение заголовков (имен столбцов) указанной таблицы"""
|
||
async with async_engine.connect() as connection:
|
||
def sync_inspect(conn):
|
||
inspector = inspect(conn)
|
||
columns = inspector.get_columns(table_name)
|
||
return [column['name'] for column in columns]
|
||
|
||
column_names = await connection.run_sync(sync_inspect)
|
||
return column_names
|
||
|
||
@classmethod
|
||
async def insert_table(cls, data: list, table_name:str = "Сollection_form_opk"):
|
||
data = data["data"]
|
||
metadata = MetaData()
|
||
async with async_engine.connect() as connection:
|
||
await connection.run_sync(metadata.reflect)
|
||
table = Table(table_name, metadata, autoload_with=connection)
|
||
async with async_session() as session:
|
||
async with session.begin():
|
||
for row in data:
|
||
stmt = insert(table).values(**row)
|
||
await session.execute(stmt)
|
||
await session.commit()
|
||
|
||
@classmethod
|
||
async def select_rows_with_filter(cls, value:str, table_name: str = "Сollection_form_opk", column_name: str = "ПОО"):
|
||
"""Выгрузка строк с фильтром по указанному столбцу и значению"""
|
||
metadata = MetaData()
|
||
async with async_engine.connect() as connection:
|
||
await connection.run_sync(metadata.reflect)
|
||
table = Table(table_name, metadata, autoload_with=connection)
|
||
|
||
async with async_session() as session:
|
||
query = select(table).where(table.c[column_name] == value)
|
||
result_proxy = await session.execute(query)
|
||
result = [row._asdict() for row in result_proxy.fetchall()]
|
||
return result |