feat:
- 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
This commit is contained in:
0
core/__init__.py
Normal file
0
core/__init__.py
Normal file
68
core/database.py
Normal file
68
core/database.py
Normal file
@@ -0,0 +1,68 @@
|
||||
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
|
||||
Reference in New Issue
Block a user