Vou enviar por partes o sistema da aplicação para minha duvida:
ENV:
DB_HOST=localhost
DB_USER=postgres
DB_PASSWORD=
DB_ADMIN_DATABASE=postgres
DB_APPLICATION_DATABASE=
Config:
from dotenv import load_dotenv
import os
class Config:
def __init__(self):
load_dotenv()
self.host = os.getenv("DB_HOST")
self.user = os.getenv("DB_USER")
self.password = os.getenv("DB_PASSWORD")
self.db_admin = os. getenv("DB_ADMIN_DATABASE")
self.db_app = os.getenv("DB_APPLICATION_DATABASE")
requerimentos = [self.host, self.user, self.password, self.db_admin, self.db_app]
if not all(requerimentos):
raise RuntimeError("Todas as variáveis do .env são obrigatórias!".format
())
db/Manual:
import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
from psycopg2 import sql
from Escola_Project.Configuraçoes.Config import Config
class ManualDataBase:
def __init__(self):
self.config = Config()
self.conn = None
def conectar(self):
try:
self.conn = psycopg2.connect(
host=self.config.host,
user=self.config.user,
password=self.config.password,
database=self.config.db_admin
)
self.conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
print("Conectado ao banco administrador com sucesso")
except psycopg2.Error as e:
print("Erro ao conectar ao banco administrador: {}".format(e))
raise
def criar_bd(self):
try:
with self.conn.cursor() as cursor:
cursor.execute(
"SELECT 1 FROM pg_database WHERE datname = %s",
(self.config.db_app,)
)
if not cursor.fetchone():
cursor.execute(
sql.SQL("CREATE DATABASE {}").format(
sql.Identifier(self.config.db_app)
)
)
print("Banco de dados '{}' criado com sucesso".format(
self.config.db_app
))
else:
print("Banco de dados '{}' já existe".format(
self.config.db_app
))
except psycopg2.Error as e:
print("Erro ao criar o banco de dados: {}".format(e))
raise
def conectar_bd_app(self):
try:
if self.conn:
self.conn.close()
self.conn = psycopg2.connect(
host=self.config.host,
user=self.config.user,
password=self.config.password,
database=self.config.db_app
)
print("Conectado ao banco da aplicação '{}'".format(
self.config.db_app
))
except psycopg2.Error as e:
print("Erro ao conectar ao banco da aplicação: {}".format(e))
raise
def teste_app(self):
try:
with self.conn.cursor() as cursor:
cursor.execute("SELECT current_database()")
db_name = cursor.fetchone()[0]
print("Banco ativo no momento: '{}'".format(db_name))
except psycopg2.Error as e:
print("Erro ao testar o banco de dados: {}".format(e))
raise
def fechar(self):
try:
if self.conn and not self.conn.closed:
self.conn.close()
print("Conexão com o banco encerrada com sucesso")
except psycopg2.Error as e:
print("Erro ao fechar a conexão: {}".format(e))
Continua.....