Criando um endpoint com dados de dois models no FastAPI
Decidi escrever esse artigo depois de passar pela necessidade de implementar um recurso GET para retornar para um cliente mobile atributos de dois models diferentes, usuário e empresa.
No backend o relacionamento entre empresa e usuário é OneToOne, onde a empresa possui em FK para usuário.
Estrutura dos Models no backend (Django)
class Usuario(Base):
django_user = models.OneToOneField(
User, on_delete=models.CASCADE, null=True,
blank=True, verbose_name="Django User",
)
nome = models.CharField("Nome", max_length=300)
email = models.EmailField("E-mail", unique=True)
telefone = models.CharField(
"Telefone", max_length=100, blank=True, null=True)
endereco = models.TextField(
"Endereço Residencial", blank=True, null=True)
class Empresa(Base):
usuario = models.ForeignKey(Usuario, on_delete=models.PROTECT)
cnpj = models.CharField("CNPJ", max_length=18, unique=True)
razao_social = models.CharField("Razão Social", max_length=300)
logo = models.ImageField(
"Logo", upload_to="usuario/empresa/logo",
blank=True, null=True)
acesso_liberado = models.BooleanField(
"Acesso Liberado", default=False)
Na camada da API (FastAPI) temos a estrutura correspondente
class Usuario(CoreBase):
__tablename__ = "usuario_usuario"
django_user_id: Mapped[Integer] = mapped_column(
ForeignKey("auth_user.id"), nullable=True, unique=True
)
nome: Mapped[String] = mapped_column(String(300), nullable=False)
email: Mapped[String] = mapped_column(
String(254), nullable=False, unique=True)
telefone: Mapped[String] = mapped_column(
String(100), nullable=True)
endereco: Mapped[String] = mapped_column(
String, nullable=True)
class Empresa(CoreBase):
__tablename__ = "usuario_empresa"
usuario_id: Mapped[UUID] = mapped_column(
ForeignKey("usuario_usuario.id"), nullable=True
)
usuario: Mapped["Usuario"] = relationship(
"Usuario", foreign_keys=[usuario_id],
backref="usuario_empresa"
)
cnpj: Mapped[String] = mapped_column(
String(18), nullable=False, unique=True)
razao_social: Mapped[String] = mapped_column(
String(300), nullable=False)
logo: Mapped[String] = mapped_column(
String(100), nullable=True)
twitter: Mapped[String] = mapped_column(
String(100), nullable=True)
acesso_liberado: Mapped[Boolean] = mapped_column(
Boolean, default=False)
A parte que merece atenção na declaração dos models na camada FastAPI é a ligação entre empresa e usuário. Como estou utilizando o SqlAlchemy a ligação entre as tabelas é feita em duas partes.
...
usuario_id: Mapped[UUID] = mapped_column(
ForeignKey("usuario_usuario.id"), nullable=True
)
usuario: Mapped["Usuario"] = relationship(
"Usuario", foreign_keys=[usuario_id], backref="usuario_empresa"
)
...
Schemas
No FastAPI trabalhamos com a estrutura de schemas para controlar, entre várias coisas, o parser e validação dos dados. Para atender o requisito da tela declarei um schema de Empresa que reuni os dados necessários para serem retornados ao cliente. No schema abaixo os campos usuario_nome, usuario_email, usuario_telefone, usuario_endereco receberão os valores do models Usuario
class EmpresaDetailBase(BaseModel):
usuario_id: Optional[UUID]
cnpj: str
razao_social: str
logo: Optional[str] = None
usuario_nome: Optional[str] = None
usuario_email: Optional[EmailStr] = None
usuario_telefone: Optional[str] = None
usuario_endereco: Optional[str] = None
Por fim, na consulta aos dados tive também que realizar alguns ajustes. O primeiro deles foi na query fazer uso do comando join do SqlAlchemy para trazer além dos dados da empresa os dados correspondentes do usuário, realizando o join por meio do Empresa.usuario, ficando assim o comando.
query = (
select(Empresa, Usuario)
.join(Empresa.usuario)
.where(Empresa.deleted.is_(False))
)
Como preciso dos dados dos dois objetos no resultado da query executei o comando via .all(), o que me traz uma lista com o primeiro elemento sendo os dados da empresa e o segundo os dados do usuário.
Com os dados em mãos realizo um for para percorrer item a item da lista, lembrando que é uma lista do tipo (Empresa, Usuario).
for _empresa in _empresas:
_nova_empresa = _empresa[0]
_novo_usuario = _empresa[1]
_empresa_detail = EmpresaDetailBase(**_nova_empresa.__dict__)
_empresa_detail.usuario_nome = _novo_usuario.nome
_empresa_detail.usuario_email = _novo_usuario.email
_empresa_detail.usuario_telefone = _novo_usuario.telefone
_empresa_detail.usuario_endereco = _novo_usuario.endereco
_itens.append(_empresa_detail.model_dump())*
* _itens é uma lista de EmpresaDetailBase
O resultado foi esse.
[
{
"usuario_id": "95c1c83c-1916-426e-b082-19156735c2bf",
"cnpj": "48.561.237/0001-78",
"razao_social": "Campos",
"logo": "https://picsum.photos/664/320",
"usuario_nome": "Pereira - ME",
"usuario_email": "[email protected]",
"usuario_telefone": "0500 037 7420",
"usuario_endereco": "Feira Moreira, Moreira Grande / PB"
},
{
"usuario_id": "84f5e113-5071-42a6-9069-31da06e5cd0a",
"cnpj": "56.304.821/0001-76",
"razao_social": "Cardoso Ferreira - ME",
"logo": "https://dummyimage.com/829x324",
"usuario_nome": "Duarte",
"usuario_email": "[email protected]",
"usuario_telefone": "+55 (051) 0749 1812",
"usuario_endereco": "Trevo Heitor Correia, Costa / GO"
},
]
Essa foi a abordagem que eu encontrei, gostaria da ajuda de vocês sugerindo melhorias na implementação.
O FastAPI tem se mostrado um ótimo framework por aplicar fortemente, na minha opinião, o conceito de Simple is better than complex.