Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Desafio de Projeto Desenvolvendo sua Primeira API com FastAPI, Python e Docker #5

Open
Maryraiane opened this issue May 17, 2024 · 0 comments

Comments

@Maryraiane
Copy link

Instalação das Dependências
bash
pip install fastapi sqlalchemy fastapi-pagination uvicorn

Criação do Modelo de Dados
Python
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship

from database import Base

class Atleta(Base):
tablename = "atletas"

id = Column(Integer, primary_key=True, autoincrement=True)
nome = Column(String(255), nullable=False)
cpf = Column(String(11), unique=True, nullable=False)
centro_treinamento = Column(String(255))
categoria = Column(String(255))

# Relacionamento com tabela de treinos (opcional)
# treinos = relationship("Treino", backref="atleta")

Configuração do Banco de Dados:
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

engine = create_engine("postgresql://user:password@host:port/database")
Base.metadata.create_all(engine)

SessionLocal = Session(autocommit=False, autoflush=False, bind=engine)

Implementação da API Assíncrona
Python
from fastapi import FastAPI, Depends, HTTPException
from fastapi_pagination import Page, add_pagination
from sqlalchemy.orm import Session

from models import Atleta
from database import SessionLocal

app = FastAPI()

@app.on_event("startup")
async def startup_event():
global session
session = SessionLocal()

@app.on_event("shutdown")
async def shutdown_event():
global session
session.close()

Obter sessão do banco de dados

def get_db(session: Session = Depends(SessionLocal)):
return session

Endpoint para buscar todos os atletas

@app.get("/atletas", response_model=Page[Atleta])
async def get_all_atletas(db: Session = Depends(get_db), limit: int | None = None, offset: int | None = None):
atletas = db.query(Atleta).order_by(Atleta.id).limit(limit).offset(offset).all()
return Page(atletas, len(atletas))

Endpoint para buscar atletas por nome e CPF

@app.get("/atletas", response_model=List[Atleta])
async def get_atleta_by_nome_cpf(nome: str | None = None, cpf: str | None = None, db: Session = Depends(get_db)):
query = db.query(Atleta)

if nome:
    query = query.filter(Atleta.nome.ilike(f"%{nome}%"))

if cpf:
    query = query.filter(Atleta.cpf == cpf)

atletas = query.all()

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

return atletas

Endpoint para cadastrar um novo atleta

@app.post("/atletas")
async def

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

1 participant