55 lines
1.4 KiB
Python
55 lines
1.4 KiB
Python
# core/database/connection_manager.py
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.engine import Engine
|
|
|
|
from actions.config.config import Config
|
|
|
|
|
|
class Firebird:
|
|
_engine: Optional[Engine] = None
|
|
|
|
@classmethod
|
|
def get_engine(cls) -> Engine:
|
|
|
|
# Obtem as configurações de banco de dados
|
|
database = Config.get('database/firebird.json')
|
|
|
|
# Se nao existir engine, cria uma nova
|
|
if cls._engine is None:
|
|
|
|
# Caminho da conexão com o banco de dados
|
|
dsn = (
|
|
f"firebird://{database.user}:"
|
|
f"{database.password}@"
|
|
f"{database.host}:"
|
|
f"{database.port}/"
|
|
f"{database.name}"
|
|
)
|
|
|
|
# Criação da Engine
|
|
cls._engine = create_engine(
|
|
dsn,
|
|
connect_args={
|
|
"charset": database.charset,
|
|
},
|
|
pool_pre_ping=database.pool.pre_ping,
|
|
pool_size=database.pool.size,
|
|
max_overflow=database.pool.max_overflow
|
|
)
|
|
|
|
# Retorna a criação da engine
|
|
return cls._engine
|
|
|
|
@classmethod
|
|
def dispose(cls):
|
|
|
|
# Verifica se existe engine
|
|
if cls._engine:
|
|
|
|
# Se existir encerra a conexão
|
|
cls._engine.dispose()
|
|
|
|
# Anula a engine
|
|
cls._engine = None
|