3
respostas

[Dúvida] Esta conforme ?. Mandei por partes o codigo ok!!!!

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.....

3 respostas

db_orm:

from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.exc import SQLAlchemyError
from Escola_Project.Configuraçoes.Config import Config

class OrmDataBase:
    base = declarative_base()

    def __init__(self):
        self.config = Config()

        self.engine = create_engine(
            "postgresql://{}:{}@{}/{}".format(
                self.config.user,
                self.config.password,
                self.config.host,
                self.config.db_app
            )
        )
        self.SessionLocal = sessionmaker(bind=self.engine)

    def get_session(self):
        db = self.SessionLocal()
        try:
            yield db
        finally:
            db.close()

    def criar_tabelas(self):
        OrmDataBase.base.metadata.create_all(self.engine)

    def testar_orm(self):
        try:
            with self.engine.connect() as conn:
                result = conn.execute(
                    text("SELECT current_database()")
                )
                db_name = result.scalar()
                print("ORM conectada no Banco: {}".format(db_name))
        except SQLAlchemyError as e:
            print("ERRO ao conectar na ORM: {}".format(e))
            raise

Models estudante:

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from Escola_Project.Core.Databases.db_orm import OrmDataBase

class Estudante(OrmDataBase.base):
    __tablename__ = "estudantes"''

    id = Column(Integer, primary_key=True, index=True)
    nome = Column(String(100), nullable=False)
    idade = Column(Integer)

    matriculas = relationship("Matricula")

Models matriculas:

from sqlalchemy import Column, Integer, String, ForeignKey
from Escola_Project.Core.Databases.db_orm import OrmDataBase

class Matricula(OrmDataBase.base):
    __tablename__ = "matriculas"

    id = Column(Integer, primary_key=True, index=True)

    estudante_id = Column(
        Integer,
        ForeignKey("estudantes.id"),
        nullable=False
    )

    nome_disciplina = Column(
        String(100),
        nullable=False
    )

schemas:

from pydantic import BaseModel

# -------- ESTUDANTE --------

class EstudanteBase(BaseModel):
    nome: str
    idade: int

class EstudanteCreate(EstudanteBase):
    pass

class EstudanteResponse(EstudanteBase):
    id: int

    class Config:
        from_attributes = True

# -------- MATRICULA --------

class MatriculasBase(BaseModel):
    estudante_id: int
    nome_disciplina: str

class MatriculaCreate(MatriculasBase):
    pass

class MatriculaResponse(MatriculasBase):
    id: int

    class Config:
        from_attributes = True

Continua.....

rotes:

from fastapi import APIRouter, HTTPException
from sqlalchemy.orm import Session

from Escola_Project.Core.Databases.db_orm import OrmDataBase
from Escola_Project.models.models_Estudante import Estudante
from Escola_Project.models.models_matriculas import Matricula
from Escola_Project.schemas.schemas import (
    EstudanteCreate,
    EstudanteResponse,
    MatriculaCreate,
    MatriculaResponse
)

router = APIRouter()
db_instance = OrmDataBase()

# ------------------ ESTUDANTES ------------------

@router.post("/estudantes", response_model=EstudanteResponse)
def criar_estudante(estudante: EstudanteCreate):
    session: Session = db_instance.SessionLocal()

    novo = Estudante(
        nome=estudante.nome,
        idade=estudante.idade
    )

    session.add(novo)
    session.commit()
    session.refresh(novo)
    session.close()

    return novo


@router.get("/estudantes", response_model=list[EstudanteResponse])
def listar_estudantes():
    session: Session = db_instance.SessionLocal()

    estudantes = session.query(Estudante).all()
    session.close()

    return estudantes


@router.get("/estudantes/{estudante_id}", response_model=EstudanteResponse)
def buscar_estudante(estudante_id: int):
    session: Session = db_instance.SessionLocal()

    estudante = session.query(Estudante).filter(
        Estudante.id == estudante_id
    ).first()

    session.close()

    if not estudante:
        raise HTTPException(status_code=404, detail="Estudante não encontrado")

    return estudante


# ------------------ MATRÍCULAS ------------------

@router.post("/matriculas", response_model=MatriculaResponse)
def criar_matricula(matricula: MatriculaCreate):
    session: Session = db_instance.SessionLocal()

    estudante = session.query(Estudante).filter(
        Estudante.id == matricula.estudante_id
    ).first()

    if not estudante:
        session.close()
        raise HTTPException(status_code=404, detail="Estudante não encontrado")

    nova = Matricula(
        estudante_id=matricula.estudante_id,
        nome_disciplina=matricula.nome_disciplina
    )

    session.add(nova)
    session.commit()
    session.refresh(nova)
    session.close()

    return nova


@router.get("/matriculas", response_model=list[MatriculaResponse])
def listar_matriculas():
    session: Session = db_instance.SessionLocal()

    matriculas = session.query(Matricula).all()
    session.close()

    return matriculas


@router.get(
    "/estudantes/{estudante_id}/matriculas",
    response_model=list[MatriculaResponse]
)
def listar_matriculas_estudante(estudante_id: int):
    session: Session = db_instance.SessionLocal()

    estudante = session.query(Estudante).filter(
        Estudante.id == estudante_id
    ).first()

    if not estudante:
        session.close()
        raise HTTPException(status_code=404, detail="Estudante não encontrado")

    matriculas = estudante.matriculas
    session.close()

    return matriculas

main:

from Escola_Project.Core.Databases.db_orm import OrmDataBase
from Escola_Project.rotes.Https_hhtp import router

app = FastAPI(
    title="Sistema Escolar",
    description="API de estudantes e matrículas",
    version="1.0.0"
)


@app.on_event("startup")
def startup_event():
    # -------- CRIA O BANCO SE NÃO EXISTIR --------
    db_manual = ManualDataBase()
    db_manual.conectar()        # conecta no banco admin (postgres)
    db_manual.criar_bd()        # cria o banco da aplicação se não existir
    db_manual.fechar()

    # -------- CRIA AS TABELAS --------
    orm = OrmDataBase()
    orm.criar_tabelas()
    orm.testar_orm()

    print("API pronta para uso!")


# -------- ROTAS --------
app.include_router(router, prefix="/api")

Oi, Samuel! Como vai?
Agradeço por compartilhar seu código com a comunidade Alura.

Ficou bem legal a separação entre Config, a criação manual do banco com psycopg2 e o uso do SQLAlchemy nas rotas.

Pontos importantes: no model Estudante, o __tablename__ = "estudantes"'' tem aspas a mais (isso quebra o model), e no .env o DB_PASSWORD e o DB_APPLICATION_DATABASE vazios vão disparar o RuntimeError do seu Config (porque você valida com all()).

Uma dica interessante para o futuro é usar Depends para gerenciar a sessão do banco e não precisar abrir e fechar manualmente em cada rota. Veja este exemplo:


from fastapi import Depends
from sqlalchemy.orm import Session

def get_db():
    db = db_instance.SessionLocal()
    try:
        yield db
    finally:
        db.close()

@router.get("/estudantes")
def listar_estudantes(db: Session = Depends(get_db)):
    return db.query(Estudante).all()

Esse código cria uma dependência que entrega a sessão (db) para a rota e garante o close() no final.

Alura Conte com o apoio da comunidade Alura na sua jornada. Abraços e bons estudos!