Martín Miranda

Creación del proyecto Sage API con FastAPI and Starlatte

__pycache__/
.idea
.env
... ...
DEBUG=False
DATABASE_URL=mssql+pymssql://user:password@database_host/db_name
... ...
from sqlalchemy import create_engine, MetaData, Table
from sqlalchemy.orm import mapper
from sqlalchemy.sql import select
from config import settings
class Persona(object):
pass
engine = create_engine(settings.database_url)
metadata = MetaData(engine)
tabla_persona = Table('tb_personas', metadata, autoload=True)
mapper(Persona, tabla_persona)
print(tabla_persona.columns)
# Ejecutar una query
conn = engine.connect()
s = select([Persona.Apellido]).where(Persona.id == 802)
res = conn.execute(s)
row = res.fetchone()
res = None if row is None else row['ColumnName']
print(res)
... ...
from sqlalchemy import Column, INTEGER, VARCHAR, ForeignKey, DATE
from sqlalchemy.orm import relationship
from database import ModelBase
class Sexo(ModelBase):
__tablename__ = 'tb_sexo'
id = Column('idSexo', INTEGER(), primary_key=True, nullable=False)
nombre = Column('Descripcion', VARCHAR(length=17), nullable=False)
abreviacion = Column('Mnemo', VARCHAR(length=1))
def __str__(self):
return f"{self.nombre}"
def __repr__(self):
return f"<Sexo: {self}>"
class Persona(ModelBase):
__tablename__ = 'tb_personas'
id = Column('idPersona', INTEGER(), primary_key=True, nullable=False)
apellido = Column('Apellido', VARCHAR(length=100), nullable=False)
nombres = Column('Nombres', VARCHAR(length=100), nullable=False)
cuil = Column('CUIL', VARCHAR(length=20))
documento = Column('Documento', INTEGER(), nullable=False)
fecha_nacimiento = Column('FechaNacimiento', DATE(), nullable=False)
domicilio = Column('Domicilio', VARCHAR(length=100))
email = Column('Email', VARCHAR(length=50))
sexo_id = Column('Sexo', INTEGER(), ForeignKey('tb_sexo.idSexo'), nullable=False)
sexo = relationship('Sexo')
es_discapacitado = Column('EsDiscapacitado', VARCHAR(length=1), nullable=False)
codigo_postal = Column('CodPostal', VARCHAR(length=20))
def __str__(self):
return f"{self.apellido}, {self.nombres}"
def __repr__(self):
return f"<Persona {self}>"
... ...
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from persona.models import Sexo, Persona
from persona.schemas import SexoSchema, PersonaSchema
router = APIRouter()
@router.get("/sexos/", response_model=List[SexoSchema])
def obtener_sexos(db: Session = Depends(get_db)):
queryset = db.query(Sexo).all()
return queryset
@router.get('/personas/', response_model=List[PersonaSchema])
def obtener_personas(
skip: int = 0,
limit: int = 50,
search: Optional[str] = None,
db: Session = Depends(get_db)):
if search:
return db.query(Persona).filter(Persona.documento == search).all()
return db.query(Persona).offset(skip).limit(limit).all()
@router.get('/personas/{persona_id}/', response_model=PersonaSchema)
def obtener_persona_por_id(persona_id: int, db: Session = Depends(get_db)):
persona = db.query(Persona).filter(Persona.id == persona_id).first()
if persona is None:
raise HTTPException(status_code=404, detail=f"No se encontró la persona con id {persona_id}")
return persona
... ...
from datetime import date
from typing import Optional
import pydantic
class SexoSchema(pydantic.BaseModel):
id: int
nombre: str
abreviacion: Optional[str] = None
class Config:
orm_mode = True
class PersonaSchema(pydantic.BaseModel):
id: int
apellido: str
nombres: str
cuil: Optional[str] = None
documento: int
fecha_nacimiento: date
domicilio: Optional[str] = None
email: Optional[str] = None
sexo: SexoSchema
es_discapacitado: str
codigo_postal: Optional[str] = None
class Config:
orm_mode = True
... ...
import sys
from pathlib import Path
from pydantic import BaseSettings
ROOT_DIR = Path(__file__).resolve().parent.parent
PROJECT_DIR = ROOT_DIR / 'project'
APPS_DIR = PROJECT_DIR / 'apps'
sys.path.append(str(APPS_DIR)) # Add imports for apps.
class Settings(BaseSettings):
database_url: str = 'example'
debug: bool = False
class Config:
env_file = str(ROOT_DIR / '.env')
settings = Settings()
... ...
from sqlalchemy import create_engine, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from config import settings
engine = create_engine(settings.database_url)
metadata = MetaData(engine)
ModelBase = declarative_base(metadata=metadata)
SessionLocal = sessionmaker(engine, autoflush=False, autocommit=False)
# Dependency
def get_db():
try:
db = SessionLocal()
yield db
finally:
db.close()
... ...
from fastapi import FastAPI
from config import settings
from persona.routers import router as persona_router
app = FastAPI(debug=settings.debug)
app.include_router(persona_router)
... ...
# Fastapi
fastapi==0.61.1
python-dotenv==0.14.0
# Database
SQLAlchemy==1.3.20
pymssql==2.1.5
\ No newline at end of file
... ...
-r base.txt
ipython==7.18.1
# server development
uvicorn==0.12.2
\ No newline at end of file
... ...