wiki-_-Guia-de-desarrollo

_ Guia de desarrollo

Guia de desarrollo

Como esta estructurado el codigo

aurapi-v2/
|-- src
| |-- shared/
| |-- feature_1/
| | |-- aplicacion/
| | |-- casos de usos
| | |-- domain/
| | | |-- interfacer/
| | | |-- entidad.py
| | | |-- ...
| | |-- infra/
| | | |-- db/
| | | |-- apis/
| | | |-- api_client/
| | | |-- ...
| |-- feature_2/
|-- tests/
| |-- e2e
| | |-- shared/
| | |-- feature_1/
| | |-- feature_2/
| |-- units
| | |-- shared/
| | |-- feature_1/
| | |-- feature_2/
|-- run_api.py

Porque tests separado de src ?

Los test no son parte del codigo productivo y tenerlos dentro de la carpeta src complica mas la lectura del codigo. Ademas tenerlo separado permite buildear una imagen ignorando la carpeta completa y asi no subir los test a produccion.

src

Porque por feature?

Cuando desarrollamos lo hacemos por feature no por capas(apis, base de datos, etc) tener esta separacion nos permite tener todo contenido aca, sin necesidad de estar desplegando carpetas con multiples archivos/carpetas de features que no nos interesa
Esto permite extraer la carpeta y poder sacar el feature al otro servicio con facilidad
La dependencia que hay entre features deberian ser minimas para aprovechar estas ventajas

Que es shared?

Todas las funciones generales como autenticacion de usuario,validacion de permisos, conexion a la base de datos, etc.

Lo ideal es que esta carpeta sea reemplazada en su mayoria por package de python, asi podemos compartir codigo con otros servicios.

run_api.py

Punto de entrada para la api.

Instancia y configura la app
Define las rutas
Instancia las dependencias 💡 Tip:

Si es necesario crear otro punto de entra como puede ser celery, definirlo en la raiz del proyecto

Clean architecture

arquitectura

Nuestra arquitectura sigue los principios de Clean Architecture, organizando el código en capas concéntricas con reglas estrictas de dependencias. Esto permite mantener un diseño modular, escalable y fácil de mantener.

En la imagen separa la ultima capa en presentacion, persitencia e infraestructura a todo esto lo llamaremos infraestructura.

Regla de Dependencias

Las flechas en el gráfico representan la regla de dependencias: Las capas más externas pueden depender de las más internas, pero nunca al revés. Esto garantiza que los detalles de implementación estén separados del núcleo de la lógica de negocio.

Dominio

Esta es la capa central y más importante, encargada de definir todo lo relacionado con la lógica de negocio.

Responsabilidades:

Entidades: Representan los objetos principales del dominio con sus atributos y comportamientos.
Funciones: Contienen la lógica de negocio pura.
Interfaces: Definen contratos (como repositorios y servicios externos) que otras capas deben implementar, pero no las implementan directamente.

Restricciones:

No puede depender de ninguna otra capa.
Debe ser completamente independiente de detalles tecnológicos o de frameworks.

Aplicacion

Esta capa se encarga de coordinar la lógica del dominio para ejecutar los casos de uso de la aplicación.

Responsabilidades:

Contiene todos los casos de uso. Al leer esta capa, debería ser claro qué funcionalidades tiene la aplicación, sin importar cómo se consumen o cómo se persiste la información.
Actúa como intermediaria entre el dominio y la infraestructura, delegando tareas mediante las interfaces definidas en el dominio.

Restricciones:

No puede saber cómo ni dónde se persisten los datos.
No puede depender de detalles de implementación de otras capas.

Infraestructura

Responsabilidades:

Implementar las interfaces definidas en el dominio.
Manejar detalles específicos como frameworks, bases de datos o librerías externas.

Ejemplo

Utilizare el login de Fieldview como ejemplo. El caso de uso es la generacion del token de fieldview

Dominio

Defino una entidad LoginUserFieldview, aca encapsulo la logica de negocio que me dice si un usuario esta logeado

# /domain/login_user_fieldview.py
@dataclass
class LoginUserFieldview:
 user_id: int
 access_token: str
 refresh_token: str
 created_at: datetime = field(default=datetime.now(timezone.utc))

 def is_access_token_valid(self) -> bool:
 """Verifica si el access_token esta en fecha

 Returns:
 bool: True si todavia se puede usar el token, False se tiene que generar uno nuevo
 """
 ...

 def is_refresh_token_valid(self) -> bool:
 """Verifica si el refresh_token esta en fecha

 Returns:
 bool: True si todavia se puede usar el token, False se tiene que generar uno nuevo
 """
 ...
 def is_valid_login(self) -> bool:
 """Si uno de los 2 token es valido esta logueado

 Returns:
 bool: True si uno de los 2 token se puede usar
 """
 ...

Ademas defino 2 interfaces, una para el repositorio de la base de datos y otra "api_client"

# /interfaces/login_fieldview_repository_abc.py
from abc import ABC, abstractmethod

from src.fieldview.domain.login_user_fieldview import LoginUserFieldview


class LoginFieldviewRepositoryABC(ABC):
 @abstractmethod
 async def save(self, login_user: LoginUserFieldview):
 pass

 @abstractmethod
 async def get(self, user_id: int) -> LoginUserFieldview | None:
 pass

 @abstractmethod
 async def delete(self, user_id: int):
 pass

# /interfaces/login_fieldview_api_client_abc.py
from abc import ABC, abstractmethod

from src.fieldview.domain.login_user_fieldview import LoginUserFieldview


class LoginFieldviewApiClientABC(ABC):
 @abstractmethod
 async def generate_token(self, user_id: int, url_redirect: str, code: str) -> LoginUserFieldview:
 pass

 @abstractmethod
 async def refresh_token(self, user_id: int, refresh_token: str) -> LoginUserFieldview | None:
 pass

 @abstractmethod
 def get_url_redirect(self, user_id: int) -> str:
 pass

 @abstractmethod
 def get_url_login(self, user_id: int) -> str:
 pass

Solo estoy definiendo los metodos publicos que luego utilizare en mi capa de aplicacion. Pero no se nada de su implementacion.

Utilizo como sub-fijo abc, para identificar que son "interfaces"

Aplicacion

Recibe en el constructor las implementaciones de las interfaces(patron Dependency injection). Esto nos permite utilizar a la capa de infraestructura sin saber de su implementacion.

# /application/login/login_user_fieldview.py
class LoginFieldview:
 def __init__(
 self,
 login_fieldview_repository: LoginFieldviewRepositoryABC,
 login_fieldview_api_client: LoginFieldviewApiClientABC,
 ):
 self._login_fieldview_repository = login_fieldview_repository
 self._login_fieldview_api_client = login_fieldview_api_client

 async def execute(self, user_id: int, code: str):
 """genera access_token y refresh_token y lo guarda en db

 Args:
 user_id (int):
 code (str): codigo de fieldview para generar token, este codigo es valido aproximadamente por 1 minuto
 """
 url_redirect = self._login_fieldview_api_client.get_url_redirect(user_id)
 login_user_fieldview = await self._login_fieldview_api_client.generate_token(user_id, url_redirect, code)

 await self._login_fieldview_repository.save(login_user_fieldview)

Infraestructura

La capa de Infraestructura implementa los detalles necesarios para conectar la lógica de negocio con tecnologías externas, siguiendo las interfaces definidas en el dominio. Para una mejor organización, la infraestructura está dividida en carpetas según el tipo de responsabilidad:

api_client:

Apis de terceros, No voy a entrar en detalle es la implementacion de LoginFieldviewApiClientABC, consume las apis de fieldview.

db: Repositorios

Los repositorios tiene que recibir en el constructor la clase implementacion de la interface DatabaseConnectionABC. De esta manera el repositorio no se encarga de la conexion a la base de datos si no que la delega a DatabaseConnectionABC. El repositorio no deberia tener logica de negocio, solo la logica de como guardar y buscar la informacion

from src.fieldview.domain.interfaces.login_fieldview_repository_abc import (
 LoginFieldviewRepositoryABC,
)
from src.fieldview.domain.login_user_fieldview import LoginUserFieldview
from src.shared.domain.interface.db_connection_abc import DatabaseConnectionABC


class LoginFieldviewRepository(LoginFieldviewRepositoryABC):
 def __init__(self, db_connection: DatabaseConnectionABC):
 self._db_connection = db_connection

 async def save(self, login_user: LoginUserFieldview):
 query = """
 INSERT INTO fieldview_login (user_id, access_token, refresh_token, created_at, deleted_at)
 VALUES (%(user_id)s, %(access_token)s, %(refresh_token)s, %(created_at)s, NULL)
 """
 await self._db_connection.execute(
 query,
 {
 "user_id": login_user.user_id,
 "access_token": login_user.access_token,
 "refresh_token": login_user.refresh_token,
 "created_at": login_user.created_at,
 },
 )

 async def get(self, user_id: int) -> LoginUserFieldview:
 query = """
 SELECT user_id, access_token, refresh_token, created_at
 FROM fieldview_login
 WHERE user_id = %(user_id)s AND deleted_at IS NULL
 """
 result = await self._db_connection.fetchone(query, {"user_id": user_id})
 if result:
 return LoginUserFieldview(**result)
 return None

 async def delete(self, user_id: int):
 query = """
 UPDATE fieldview_login
 SET deleted_at = CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
 WHERE user_id = %(user_id)s
 """
 await self._db_connection.execute(query, {"user_id": user_id})

Apis:

Apis que expone el proyecto

Todas las apis relacionadas con login de fieldview, ver manejo de dependencias

# /infra/apis/login_fieldview_api.py
from fastapi import APIRouter, Depends

from src.fieldview.application.login.login_user_fieldview import LoginFieldview
from src.shared.apis.responses import response_json
from src.shared.containers import AppContainer

login_field_view_router = APIRouter()


@login_field_view_router.get("/login/{user_id}")
async def login_fieldview(
 user_id: int,
 code: str,
 login_use_case: LoginFieldview = Depends(lambda: AppContainer.fieldview_container.login_fieldview_use_case()),
):
 await login_use_case.execute(user_id, code)

Incluir en un router general todas las apis relacionadas con fieldview

# /infra/apis/__init__.py
from fastapi import APIRouter

from .login_fieldview_api import login_field_view_router

fieldview_router = APIRouter()

fieldview_router.include_router(login_field_view_router, prefix='/auth', tags=['Fieldview-Login'])

# run_api.py
# Agregar a la app el router de fieldview
app.include_router(fieldview_router, prefix='/fieldview')

Dependencias

Utilizo la libreria dependency-injector, Esto se podria manejar con la dependecias de fast api, la decision de usar otra libreria es indenpendizar el framework web de las dependencias ya que si tenemos un celery o otro punto de entrada deberiamos reescribirlas dependencia o usar fastapi.

https://pypi.org/project/dependency-injector/

Se maneja por contenedores, Dentro de shared esta el contenedor principal, este mismo tiene un contenedor general para todos los features y cada feature debera crear su contenedor y agregarlo al principal

# src/shared/containers.py
from dependency_injector import containers, providers

from src.fieldview.container_fieldview import FieldviewContainer
from src.shared.infrastruture.db.db_postgresql_connection import PostgresqlConnection
from src.shared.settings import settings


class InfraContainer(containers.DeclarativeContainer):
 """Contenedor para dependencias transversales."""

 db_connection = providers.Singleton(
 PostgresqlConnection,
 user=settings.pg_user,
 password=settings.pg_password,
 host=settings.pg_host,
 port=settings.pg_port,
 database=settings.pg_database,
 min_size=settings.pg_min_size,
 max_size=settings.pg_max_size,
 )


class AppContainer(containers.DeclarativeContainer):
 """Contenedor principal con dependencias modulares."""

 infra_container: InfraContainer = providers.Container(InfraContainer)

 fieldview_container: FieldviewContainer = providers.Container(
 FieldviewContainer, db_connection=infra_container.db_connection
 )

Que de deberia definir en los contenedores?

Las implementaciones de las interfaces de repositorios, api_client, etc.
Casos de usos

Ver cual es el providers que necesito para cada caso (Factory, Singleton, Dependency, Container)

Ejemplo de field view

from dependency_injector import containers, providers

...


class FieldviewContainer(containers.DeclarativeContainer):
 """Contenedor para las dependencias del módulo Fieldview."""

 # Viene como dependencia de otro contenedor
 db_connection = providers.Dependency()

 # Repos
 login_fieldview_repository = providers.Factory(LoginFieldviewRepository, db_connection=db_connection)

 # API Clients
 operation_fieldview_api_client = providers.Factory(FieldviewOperationApiClient)
 login_fieldview_api_client = providers.Factory(LoginFieldviewApiClient)

 # Use cases
 # Login
 generate_url_use_case = providers.Factory(
 GenerateUrlLoginFieldview,
 login_fieldview_repository=login_fieldview_repository,
 login_fieldview_api_client=login_fieldview_api_client,
 )
 login_fieldview_use_case = providers.Factory(
 LoginFieldview,
 login_fieldview_repository=login_fieldview_repository,
 login_fieldview_api_client=login_fieldview_api_client,
 )
 get_token_fieldview_use_case = providers.Factory(
 get_token_fieldview.GetTokenFieldview,
 login_fieldview_repository=login_fieldview_repository,
 login_fieldview_api_client=login_fieldview_api_client,
 )

# run_api.py
# Iniciar contenedores
container = AppContainer()
container.init_resources()

Como utilizar estas dependencias en Fast api

from fastapi import Depends
@login_field_view_router.get("/login/{user_id}")
async def login_fieldview(
 user_id: int,
 code: str,
 login_use_case: LoginFieldview = Depends(lambda: AppContainer.fieldview_container.login_fieldview_use_case()),
):
 await login_use_case.execute(user_id, code)

Como quedaria la ruta sin utilizar dependencias NO REALIZAR ASI, SOLO EJEMPLO PARA VER LA VENTAJAS

from fastapi import Depends
@login_field_view_router.get("/login/{user_id}")
async def login_fieldview(
 user_id: int,
 code: str,
 # login_use_case: LoginFieldview = Depends(lambda: AppContainer.fieldview_container.login_fieldview_use_case()),
):
 db = PostgresqlConnection(
 user=settings.pg_user,
 password=settings.pg_password,
 host=settings.pg_host,
 port=settings.pg_port,
 database=settings.pg_database,
 min_size=settings.pg_min_size,
 max_size=settings.pg_max_size
 )
 login_fieldview_repository = LoginFieldviewRepository(db),
 
 login_fieldview_api_client = LoginFieldviewApiClient()
 login_use_case = LoginFieldview(
 login_fieldview_repository=login_fieldview_repository,
 login_fieldview_api_client=login_fieldview_api_client
 )
 await login_use_case.execute(user_id, code)

Ventajas:

Centraliza la creacion de las instancias
Es facil reemplazar las dependencias para generar pruebas controladas (Ver en documentacion de test)

Como proteger una api

Los 3 posibles login

# /src/shared/infra/dependence.py
from src.shared.apis.authorization import UserDependency

# Dependencias para validar login y obtener el usuario
require_login = UserDependency() # No permitido extensiones
require_login_apps_only = UserDependency("apps_only") # Solo Extensiones
require_login_open_api = UserDependency("open_api") # Para todos

Luego en la definicion de la ruta

@router.get("")
async def xxx(
 user: UserRequest = Depends(require_login)
):
 ...

reemplazar require_login por el tipo de login necesario si es abierto a extensiones o no

Con esto ya estamos validando que este logueado y tendremos un objeto con la informacion del usuario

@dataclass
class UserRequest:
 auth_token: str
 developer_id: int | None
 ext_id: int | None
 ext_ver_id: int | None
 locale: str
 login_id: int
 src: str
 user_id: int
 workspace_id: int | None

Validacion de schema

Para la validacion de datos usamos la libreria Pydantic, que tiene integracion con Fastapi tanto para atajar errores de validacion de datos como la autodocumentacion.

Para ampliar sobre el tema ver tutorial de fastapi, Tiene varias secciones para ver como recibir parametros por query_params, body, path, etc. Y como se integra con swagger y validacion de datos

https://fastapi.tiangolo.com/tutorial/path-params/

Para la respuesta de la api usar la funcion, donde se podra agregar encabezados y todo lo necesario para la respuesta de todas las apis

# /src/shared/apis/responses.py
def response_json(data: BaseModel | dict, status_code: int = status.HTTP_200_OK):
 ...

Variables de entorno

Utilizo la libreria pydantic_settings, para tener las variables de entorno, esto te permite validaciones de tipo de datos, setear valores default, etc. Si falta alguna variable fallara la creacion de la instacia

https://docs.pydantic.dev/latest/concepts/pydantic_settings/

Hay un setting general, aca tiene que ir variables comunes y obligatorias, ya que si falta alguna no se ejecutara la app.

# /src/shared/settings.py
from pydantic import ConfigDict
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
 # General
 log_level: str = "INFO"

 # Base de datos
 pg_host: str
 pg_user: str
 pg_database: str
 pg_password: str
 pg_port: str = "5432"
 pg_min_size: int = 1
 pg_max_size: int = 10

 # JWT
 auth_jwt_key: str
 auth_domain: str = "com.auravant.auth"

 url_outside: str

 model_config = ConfigDict(
 env_file=".env",
 extra="allow", # Permitir entradas adicionales en el archivo .env
 )


settings = Settings()

Luego cada features define sus propias variables en su propio archivo de settings.

Errores

# run_api.py
register_exception_handlers(app)

# /src/shared/infra/exception_handler.py 
from fastapi import FastAPI, Request, status

from src.shared.apis.responses import response_json
from src.shared.application.logger import Logger
from src.shared.domain.exception_domain import DomainError
from src.shared.infra.exception_infra import (
 AuraHTTPException,
 InfraError,
)


async def domain_error_handler(request: Request, exc: DomainError):
 return response_json(
 {
 "code": exc.code,
 "msg": exc.message,
 "data": exc.data,
 },
 status_code=status.HTTP_400_BAD_REQUEST,
 )


async def infra_error_handler(request: Request, exc: InfraError):
 logger = Logger()
 logger.log_error("Error en capa de infraestructura:", exc)
 return response_json(
 {
 "code": exc.code,
 },
 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
 )


async def http_error_handler(request: Request, exc: AuraHTTPException):
 return response_json(
 {
 "code": exc.code,
 "msg": exc.msg,
 },
 status_code=exc.status_code,
 )


def register_exception_handlers(app: FastAPI):
 app.add_exception_handler(DomainError, domain_error_handler)
 app.add_exception_handler(InfraError, infra_error_handler)
 app.add_exception_handler(AuraHTTPException, http_error_handler)

Las excepciones del tipo Exception las ataja fastapi

Excepcion en capa de dominio o aplicacion, se devolvera como respuesta (code, message, data)

class DomainError(Exception):
 """Excepción base para errores de dominio."""

 code: str = None
 message: str = None
 data: dict = None

 def __init__(self, code: str = None, message: str = None, data: dict = None):
 super().__init__(message)
 self.code = code or self.code
 self.message = message or self.message
 self.data = data or self.data

Excepciones de infra, solo se devuelve code, message y data poner la mayor informacion para poder debugear

class InfraError(Exception):
 """Excepción base para errores de infraestructura.

 Al cliente solo me mostrar data.
 """

 def __init__(self, code: str, message: str, data: dict = None):
 super().__init__(message)
 self.code = code
 self.message = message
 self.data = data

Excepcion con status_code, esta excepcion solo se puede usar en capa de infra. Ya que aplicacion y dominio no saben nada de http

class AuraHTTPException(Exception):
 code = ''
 msg = ''
 status_code = 500
 data: dict = None

 def __init__(self, code: str = None, msg: str = None, data: dict = None):
 self.code = code or self.code
 self.msg = msg or self.msg
 self.data = data or self.data

_ Guia de desarrollo

Guia de desarrollo​

Porque tests separado de src ?​

src​

Porque por feature?​

Que es shared?​

run_api.py​

Clean architecture​

Regla de Dependencias​

Dominio​

Aplicacion​

Infraestructura​

Ejemplo​

Dominio​

Aplicacion​

Infraestructura​

api_client:​

db: Repositorios​

Apis:​

Dependencias​

Como proteger una api​

Validacion de schema​

Variables de entorno​

Errores​

Guia de desarrollo

Porque tests separado de src ?

src

Porque por feature?

Que es shared?

run_api.py

Clean architecture

Regla de Dependencias

Dominio

Aplicacion

Infraestructura

Ejemplo

Dominio

Aplicacion

Infraestructura

api_client:

db: Repositorios

Apis:

Dependencias

Como proteger una api

Validacion de schema

Variables de entorno

Errores