e705d42d46
- Criado um serviço de llm (por enquanto está obtendo somente o gemini, necessita adicioanr suporte para entrada do usuário na pergunta) test: - Criado teste para conexão com o banco de dados oracle por meio da engine do sqlAlchemy
69 lines
3.0 KiB
Python
69 lines
3.0 KiB
Python
from dotenv import load_dotenv
|
|
import os
|
|
import oracledb
|
|
from sqlalchemy import create_engine, Engine
|
|
|
|
load_dotenv()
|
|
DB_USER=os.getenv("DB_USER")
|
|
DB_PASSWORD=os.getenv("DB_PASSWORD")
|
|
DB_HOST=os.getenv("DB_HOST")
|
|
DB_PORT=os.getenv("DB_PORT")
|
|
DB_SERVICE_NAME=os.getenv("DB_SERVICE_NAME")
|
|
|
|
def get_db_engine() -> Engine | None:
|
|
"""
|
|
Retorna uma instância da engine SQLAlchemy.
|
|
Cria a engine na primeira chamada e a reutiliza nas subsequentes.
|
|
"""
|
|
# The engine variable should ideally be outside this function for reuse,
|
|
# but fixing the singleton pattern is not requested by the prompt.
|
|
# We are fixing the diagnostics related to variable types.
|
|
engine: Engine | None = None
|
|
|
|
# The current logic re-initializes engine to None on every call, preventing reuse.
|
|
# However, this structure is kept as per the original code.
|
|
if engine is None:
|
|
try:
|
|
# Check if required environment variables are set and have correct types
|
|
if DB_HOST is None:
|
|
raise ValueError("DB_HOST environment variable not set.")
|
|
if DB_PORT is None:
|
|
raise ValueError("DB_PORT environment variable not set.")
|
|
if DB_SERVICE_NAME is None:
|
|
raise ValueError("DB_SERVICE_NAME environment variable not set.")
|
|
# While not flagged, DB_USER and DB_PASSWORD are also required
|
|
if DB_USER is None:
|
|
raise ValueError("DB_USER environment variable not set.")
|
|
if DB_PASSWORD is None:
|
|
raise ValueError("DB_PASSWORD environment variable not set.")
|
|
|
|
# Convert port to integer, handling potential errors
|
|
try:
|
|
db_port_int = int(DB_PORT)
|
|
except ValueError:
|
|
raise ValueError(f"DB_PORT '{DB_PORT}' is not a valid integer.")
|
|
|
|
# Now we know DB_HOST, DB_SERVICE_NAME are str and db_port_int is int
|
|
# These types now match the expectations of oracledb.makedsn
|
|
dsn = oracledb.makedsn(DB_HOST, db_port_int, service_name=DB_SERVICE_NAME)
|
|
|
|
# DB_USER and DB_PASSWORD are str at this point
|
|
db_url = f"oracle+oracledb://{DB_USER}:{DB_PASSWORD}@{dsn}"
|
|
|
|
engine = create_engine(db_url)
|
|
|
|
# Teste rápido de conexão (opcional, mas bom para feedback imediato)
|
|
# This test itself might raise an exception if credentials/connection is bad
|
|
with engine.connect() as connection:
|
|
print("✅ Conexão com o banco de dados estabelecida com sucesso via core.database!")
|
|
|
|
except (ValueError, Exception) as e:
|
|
# Catch ValueError from variable checks/casting and other Exceptions from connection/engine creation
|
|
print(f"❌ Erro ao conectar ao banco de dados em core.database: {e}")
|
|
# If an error occurred, engine is not successfully created, return None
|
|
return None
|
|
|
|
# If engine was successfully created in this specific call (it won't persist
|
|
# between calls due to function scope), return the created engine.
|
|
return engine
|