ADUANASOFT/Verificacion_EFOS_Backend
10
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ajuste de cruds incompleto

Browse Source
This commit is contained in:
gerardoe
2026-04-06 16:06:31 -07:00
parent 61d386a7c7
commit 290a32364d
75 changed files with 3574 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
README.md Normal file
View File

@@ -0,0 +1,26 @@
# version 1 de Verfifca efos backend
SERVICIOS INTERNOS =================================
* GENERACION DE CRUDS APARTIR DE TEMPLATES
Entrando desde la consola a /verfica_Efos_back y ejecutando `python generator.py`,
generas un archivo "nombre_router.py" y "nombre_service.py" que gracias al codigo de `generator.py`, lee el model importado en la parte de 'get_all_models()'.
una vez importado el model, lo lee, lo pasa por la logica en la que se descratan posibles erores y segenera en la ruta
`output\generated\routers` y `output\generated\services` los templates pre armados de los cruds listos para copiar y pegar segun sea el caso.
A fecha del 06/04/26- la primera version parece funconar correctamente y solo cuesta ajustar los cruds a necesidad.
* arquitectura extrana*
el primer caso de uso de una arquitectura de modulos extendidos que engloban por cada una de las entidades.
(se entiende por entidades, cada uno de las partes que funcionanar en la logica de las bases de datos: por ejemplo Clientes-Clients, usuarios-users)
tiene de manera interna cada uno de los procesos necesarios, para la facilidad de mantenimiento. En cada uno de los modulos
`Junta el 31 de marzo 2026`
- puntos no claros de uso Diot
- se pidio que no se agregara multiempresas o multitenant, aunque hay la posibilida de hacerlo, asi como el que se pudiera facilitar la actualizacion de las listas de efos, edos.

BIN
__pycache__/database.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/core/__pycache__/dependencies.cpython-311.pyc Normal file
View File

Binary file not shown.

25
app/core/dependencies.py
View File

@@ -0,0 +1,25 @@
from sqlalchemy.orm import Session
from app.modules.clients.models import Client
def fiscal_number_exists(db: Session, fiscal_number: str) -> bool:
"""Verify if a client exist in db"""
if not fiscal_number:
return False
exists = db.query(Client).filter(
Client.fiscal_number == fiscal_number,
).first() is not None
return exists
def email_exist(db: Session, email: str) -> bool:
if not email:
return False
exist = db.query(Client).filter(
Client.email == email
).first() is not None
return exist

47
app/modules/affidavit_logs/schema.py
View File

@@ -0,0 +1,47 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum
class Status(Enum):
COMPLETED = "completed"
PARTIAL = "partial"
INITIAL = "initial"
NO_PROCECED = "no_proceced"
INVALID = "invalid"
class Operation(Enum):
DELETED ="deleted"
CREATED ="created"
UPDATED ="updated"
TRIED ="tried"
MOST ="mosted"
LESS ="less"
MINUS ="minus"
CANCELED ="canceled"
class AffidavitCreate(BaseModel):
operation: Operation = Operation.TRIED
user_id: Optional[int] = None
description: str
adjustment: Optional[str] = None
client_id: Optional[int] = None
status: Status = Status.INITIAL
sing: str = Field(..., description="Firma digital del usuario tomada desde service")
class AffidavitUpdate(AffidavitCreate):
pass
class AffidavitResponse(AffidavitCreate):
created_at: datetime
updated_at: datetime
deleted_at: datetime
class messageResponse(BaseModel):
message : str

BIN
app/modules/branches/__pycache__/route.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/modules/branches/__pycache__/schema.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/modules/branches/__pycache__/service.cpython-311.pyc Normal file
View File

Binary file not shown.

118
app/modules/branches/route.py
View File

@@ -0,0 +1,118 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.modules.users.models import Users
# Modelo base de usuario
from app.modules.branches.models import Branches
from app.modules.branches.schema import BrancheResponse, BranchesCreate, BrancheUpdate, MessageResponse
from app.modules.branches.service import BranchesService
router = APIRouter(prefix="/branches", tags=["branches"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=BrancheResponse, status_code=status.HTTP_201_CREATED)
async def create_branches(
data: BranchesCreate,
db: Session = Depends(get_db),
current_user: Users = Depends(get_current_user)
):
"""Create a branches - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create branches")
try:
result = BranchesService.create_branches(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[BrancheResponse])
def get_branchess(
db: Session = Depends(get_db),
current_user: Users = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return BranchesService.get_branchess(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=BrancheResponse)
def get_branches_by_id(
branches_id: int,
db: Session = Depends(get_db),
current_user: Users = Depends(get_current_user)
):
"""Get branches by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = BranchesService.get_branches(db=db, branches_id=branches_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=MessageResponse)
async def update_branches(
branches_id: int,
data: BrancheUpdate,
db: Session = Depends(get_db),
current_user: Users = Depends(get_current_user)
):
"""Update existing branches - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update branches")
try:
result = BranchesService.update_branches(
db=db,
branches_id=branches_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_branches(
branches_id: int,
db: Session = Depends(get_db),
current_user: Users = Depends(get_current_user)
):
"""Delete branches by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete branches")
try:
result = BranchesService.delete_branches(db=db, branches_id=branches_id, current_user=current_user)
return {"message": "branches deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

25
app/modules/branches/schema.py
View File

@@ -0,0 +1,25 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum
class BranchesCreate(BaseModel):
direccion: str
cp: int
is_phisical: bool
location_id: int
is_active: bool = True
client_id: Optional[int] = None
user_id: Optional[int] = None
class BrancheUpdate(BranchesCreate):
pass
class BrancheResponse(BranchesCreate):
created_at: datetime
updated_at: Optional[datetime] = None
class MessageResponse(BaseModel):
massage : str

65
app/modules/branches/service.py Normal file
View File

@@ -0,0 +1,65 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.branches.models import Branches
from app.modules.branches.schema import BranchesCreate, BrancheUpdate, BrancheResponse, MessageResponse
class BranchesService:
@staticmethod
def create_branches(db: Session, data: BranchesCreate):
new_branches = Branches(**data.dict())
db.add(new_branches)
db.commit()
db.refresh(new_branches)
return new_branches
@staticmethod
def get_branches(db: Session, branches_id: int, current_user):
branches = db.query(Branches).filter(Branches.id == branches_id).first()
if not branches:
raise ValueError("branches no encontrado")
return branches
@staticmethod
def update_branches(db: Session, branches_id: int, data: BrancheUpdate, current_user):
branches = db.query(Branches).filter(Branches.id == branches_id).first()
if not branches:
raise ValueError("branches no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este branches")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(branches, key, value)
db.commit()
db.refresh(branches)
return branches
@staticmethod
def delete_branches(db: Session, branches_id: int, current_user):
branches = db.query(Branches).filter(Branches.id == branches_id).first()
if not branches:
raise ValueError("branches no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este branches")
try:
branches.is_active = False
branches.deleted_at = datetime.utcnow()
branches.deleted_by = current_user.id
db.commit()
db.refresh(branches)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el branches: {e}")
return {"message": "branches eliminado correctamente"}

BIN
app/modules/clients/__pycache__/models.cpython-311.pyc
View File

Binary file not shown.

BIN
app/modules/clients/__pycache__/repository.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/modules/clients/__pycache__/route.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/modules/clients/__pycache__/schema.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/modules/clients/__pycache__/service.cpython-311.pyc Normal file
View File

Binary file not shown.

1
app/modules/clients/models.py
View File

@@ -74,3 +74,4 @@ class Client(Base):
return f"<Client(id={self.id}, rfc={self.rfc}, razon_social={self.razon_social})>"

7
app/modules/clients/repository.py
View File

@@ -0,0 +1,7 @@
from app.core.baseRepository import BaseRepository
from app.modules.clients.models import Client
from sqlalchemy.orm import Session
class ClientRepository(BaseRepository[Client]):
def __init__(self, db: Session):
super().__init__(db, Client)

85
app/modules/clients/route.py
View File

@@ -0,0 +1,85 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from app.core.auth import get_current_user
from app.modules.clients.models import Client
from app.modules.clients.repository import ClientRepository
from app.modules.clients.schema import ClientCreate, ClientUpdate, ClientResponse
from app.modules.clients.service import ClientService
from app.pipelines.crud import CreatePipe, ReadPipe, UpdatePipe, SoftDelete, ListPipe
from typing import List
router = APIRouter(prefix="/clients", tags=["clients"])
@router.post("/", response_model=ClientResponse)
async def create_client(
client_data: ClientCreate,
db: Session = Depends(get_db),
current_user: dict = Depends(get_current_user)
):
result = await ClientService.create_client(db, client_data, current_user)
if result["success"]:
return result["data"]
raise HTTPException(status_code=400, detail=result["error"])
@router.get("/{client_id}", response_model=ClientResponse)
async def get_client(
client_id: int,
db: Session = Depends(get_db),
current_user: dict = Depends(get_current_user)
):
repo = ClientRepository(db)
pipeline = ReadPipe(db, current_user, Client, repo)
pipeline.set_input_data({"id": client_id})
result = await pipeline.execute()
if result["success"]:
return result["data"]
else:
raise HTTPException(status_code=404, detail=result["error"])
@router.put("/{client_id}", response_model=ClientResponse)
async def update_client(
client_id: int,
client_data: ClientUpdate,
db: Session = Depends(get_db),
current_user: dict = Depends(get_current_user)
):
repo = ClientRepository(db)
pipeline = UpdatePipe(db, current_user, Client, repo)
pipeline.set_input_data({"id": client_id, **client_data.dict(exclude_unset=True)})
result = await pipeline.execute()
if result["success"]:
return result["data"]
else:
raise HTTPException(status_code=400, detail=result["error"])
@router.delete("/{client_id}")
async def delete_client(
client_id: int,
db: Session = Depends(get_db),
current_user: dict = Depends(get_current_user)
):
repo = ClientRepository(db)
pipeline = SoftDelete(db, current_user, Client, repo)
pipeline.set_input_data({"id": client_id})
result = await pipeline.execute()
if result["success"]:
return result["data"]
else:
raise HTTPException(status_code=400, detail=result["error"])
@router.get("/", response_model=List[ClientResponse])
async def list_clients(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db),
current_user: dict = Depends(get_current_user)
):
repo = ClientRepository(db)
pipeline = ListPipe(db, current_user, Client, repo)
pipeline.set_input_data({"skip": skip, "limit": limit})
result = await pipeline.execute()
if result["success"]:
return result["data"]
else:
raise HTTPException(status_code=400, detail=result["error"])

37
app/modules/clients/schema.py
View File

@@ -0,0 +1,37 @@
from pydantic import BaseModel, EmailStr
from typing import Optional
from app.modules.clients.models import Medio, Tercero, Operacion
class ClientBase(BaseModel):
rfc: str
email: EmailStr
short_name: Optional[str] = None
razon_social: str
fiscal_number: str
cellphone: str
medio: Medio = Medio.MANUAL
third_type: Tercero = Tercero.GLOBAL
operation_type: Operacion = Operacion.OTROS
is_foreign: bool = False
user_id: int
location_id: Optional[int] = None
class ClientCreate(ClientBase):
pass
class ClientUpdate(BaseModel):
short_name: Optional[str] = None
razon_social: Optional[str] = None
fiscal_number: Optional[str] = None
cellphone: Optional[str] = None
medio: Optional[Medio] = None
third_type: Optional[Tercero] = None
operation_type: Optional[Operacion] = None
is_foreign: Optional[bool] = None
location_id: Optional[int] = None
class ClientResponse(ClientBase):
id: int
class Config:
from_attributes = True

49
app/modules/clients/service.py Normal file
View File

@@ -0,0 +1,49 @@
from app.modules.clients.schema import ClientCreate
from app.modules.clients.models import Client
#
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core.dependencies import fiscal_number_exists, email_exist
class ClientService:
def __init__(self):
pass
@staticmethod
def create_client(db:Session, client_data: ClientCreate) -> Client:
if fiscal_number_exists(db, client_data.get("fiscal_number")):
raise ValueError("el numero fiscal ya existe")
if email_exist(db, client_data.get("email")):
raise ValueError("este email ya existe")
new_client = Client(
rfc = client_data.get("rfc"),
email = client_data.get("email"),
short_name = client_data.get("short_name"),
razon_social = client_data.get("razon_social"),
fiscal_number = client_data.get("fiscal_number"),
cellphone = client_data.get("cellphone"),
medio = client_data.get("medio"),
third_type = client_data.get("third_type"),
operation_type = client_data.get("operation_type"),
is_foreign = client_data.get("is_foreign", False),
user_id = client_data.get("user_id"),
location_id = client_data.get("location_id"),
)
try:
db.add(new_client)
db.commit()
db.refresh(new_client)
except IntegrityError as e:
db.rollback()
raise ValueError(f"Error al crear el cliente: {e}")
return new_client

20
app/modules/coments/schema.py
View File

@@ -0,0 +1,20 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum
class CommentCreate(BaseModel):
texto: str = Field(..., max(380), min(30))
user_id: Optional[int]
feed_id: Optional[int]
is_active: bool = True
class Commentupdate(CommentCreate):
pass
class CommentResponse(CommentCreate):
created_at: datetime
cretaed_by: int
class MessageResponse(BaseModel):
message: str

23
app/modules/configuration/schema.py
View File

@@ -0,0 +1,23 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum
class ConfigCreate(BaseModel):
smtp_host: Optional[str] = None
smtp_port: Optional[str] = None
smtp_user: Optional[str] = None
smtp_password: Optional[str] = None
is_active: bool = True
class Commentupdate(ConfigCreate):
pass
class ConfigResponse(ConfigCreate):
created_at: datetime
last_send: datetime
cretaed_by: int
class MessageResponse(BaseModel):
message: str

6
app/modules/credits/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/edos/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/efos/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/entreprise/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/feed/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/files/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/interactions/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/invoices/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/license/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/location/service.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/moves/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/reports/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/suppliers/service.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/taxes/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

6
app/modules/taxpayers/schema.py
View File

@@ -0,0 +1,6 @@
from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
#
import re
from enum import Enum

BIN
app/modules/users/__pycache__/route.cpython-311.pyc
View File

Binary file not shown.

BIN
app/pipelines/__pycache__/base.cpython-311.pyc
View File

Binary file not shown.

BIN
app/pipelines/__pycache__/crud.cpython-311.pyc
View File

Binary file not shown.

7
app/pipelines/base.py
View File

@@ -23,7 +23,6 @@ class Pipeline(Generic[InputT, OutputT]):
self._input_data: Any = None
self._schema = None
self._permission: Optional[str] = None
self._affidavit: Optional[Any] = None
def set_input_data(self, data: Any) -> 'Pipeline':
"""Define los datos de entrada"""
@@ -40,11 +39,6 @@ class Pipeline(Generic[InputT, OutputT]):
self._permission = permission
return self
def with_affidavit(self, affidavit: Any) -> 'Pipeline':
"""Define el certificador legal"""
self._affidavit = affidavit
return self
def add_step(self, name: str, func: Callable) -> 'Pipeline':
"""Agrega un paso al pipeline"""
self._steps.append({"name": name, "func": func})
@@ -67,7 +61,6 @@ class Pipeline(Generic[InputT, OutputT]):
context["input_data"] = self._input_data
context["schema"] = self._schema
context["permission"] = self._permission
context["affidavit"] = self._affidavit
result = context

39
app/pipelines/crud.py
View File

@@ -7,7 +7,7 @@ from app.core.auth import CurrentUser
class CreatePipe(Pipeline):
"""Pipe prefab to create enititys"""
"""Pipe prefab to create entities"""
def __init__(self, db: Session, user: CurrentUser, model: Type, repository):
super().__init__(db, user, f"CREATE_{model.__tablename__.upper()}")
@@ -60,9 +60,7 @@ class CreatePipe(Pipeline):
if ctx.get("entity"):
ctx["output"] = {c.name: getattr(ctx["entity"], c.name)
for c in self.model.__table__.columns}
print(f"FORMAT: output creado con {len(ctx['output'])} campos")
else:
print("FORMAT: No hay entity para formatear")
ctx["output"] = None
return ctx
@@ -79,11 +77,11 @@ class ReadPipe(Pipeline):
def _build(self):
self.add_step("validate_id", self._validate_id)
self.add_step("check_permissions", self._check_permissions)
self.add_step("execurte", self._execute)
self.add_step("execute", self._execute)
self.add_step("format_output", self._format)
async def _validate_id(self, ctx):
input_data = ctx["input"].raw_data
input_data = ctx.get("input_data", {})
entity_id = input_data.get("id")
if not entity_id:
raise ValueError("se requiere ID")
@@ -125,14 +123,18 @@ class UpdatePipe(Pipeline):
self.add_step("format_output", self._format)
async def _validate(self, ctx):
data = ctx["input"].get_validated()
ctx["validated_data"] = data.dict() if data else {}
input_data = ctx.get("input_data", {})
if ctx.get("schema"):
validated = ctx["schema"](**input_data)
ctx["validated_data"] = validated.dict()
else:
ctx["validated_data"] = input_data
return ctx
async def _validate_id(self, ctx):
entity_id = ctx["validated_data"].get("id")
if not entity_id:
raise ValueError("Se rqeuiere ID")
raise ValueError("Se requiere ID")
ctx["entity_id"] = entity_id
return ctx
@@ -151,10 +153,8 @@ class UpdatePipe(Pipeline):
async def _format(self, ctx):
ctx["output"] = {
"id": ctx["entity"].id,
"message": f"{self.model.__name__} actualizado exitosamente"
}
ctx["output"] = {c.name: getattr(ctx["entity"], c.name)
for c in self.model.__table__.columns}
return ctx
class SoftDelete(Pipeline):
@@ -168,7 +168,6 @@ class SoftDelete(Pipeline):
self._build()
def _build(self):
self.add_step("validate", self._validate)
self.add_step("validate_id", self._validate_id)
self.add_step("check_permissions", self._check_permissions)
self.add_step("check_already_deleted", self._check_already_deleted)
@@ -177,7 +176,7 @@ class SoftDelete(Pipeline):
self.add_step("format_output", self._format)
async def _validate_id(self, ctx):
input_data = ctx["input"].raw_data
input_data = ctx.get("input_data", {})
entity_id = input_data.get("id")
if not entity_id:
raise ValueError("se requiere ID")
@@ -222,14 +221,16 @@ class SoftDelete(Pipeline):
async def _format(self, ctx):
ctx["output"]["message"] = f"{self.model.__name__} eliminado corectamente"
ctx["output"]["deleted_at"] = ctx["entity"].deleted_at.isoformat()
ctx["output"]["deleted_by"] = ctx["entity"].delted_by
ctx["output"] = {
"id": ctx["entity"].id,
"message": f"{self.model.__name__} eliminado correctamente",
"deleted_at": ctx["entity"].deleted_at.isoformat() if ctx["entity"].deleted_at else None,
"deleted_by": ctx["entity"].deleted_by
}
return ctx
class ListPipe(Pipeline):
"""Pipeline prefab to list entitys"""
"""Pipeline prefab to list entities"""
def __init__(self, db: Session, user: CurrentUser, model: Type, repository):
super().__init__(db, user, f"LIST_{model.__tablename__.upper()}")

179
generator.py Normal file
View File

@@ -0,0 +1,179 @@
""" Generate service files "service and route", CRUD.
READ models on a modules, and generate code usin template Jija2
"""
import os
from pathlib import Path
from typing import Dict, List, Any
# #
from jinja2 import Environment, FileSystemLoader
from sqlalchemy import inspect, Table, Column
from sqlalchemy.engine import create_mock_engine
#####
from database import Base
import importlib
import enum
class CodeGenerator:
def __init__(self, templates_dir: str = "templates", output_dir: str = "generated"):
self.templates_dir = Path(templates_dir)
self.output_dir = Path(output_dir)
self.env = Environment(loader=FileSystemLoader(templates_dir))
# create directory of output
self.output_dir.mkdir(parents=True, exist_ok=True)
(self.output_dir / "services").mkdir(exist_ok=True)
(self.output_dir / "routes").mkdir(exist_ok=True)
(self.output_dir / "schemas").mkdir(exist_ok=True)
def get_model_metadata(self, model_class) -> Dict[str, Any]:
"""" Extract metadata from SQLAlchemy """
if not hasattr(model_class, '__table__'):
print(f"{model_class.__name__} no tiene __table__, jumping")
return None
fields = []
required_fields = []
unique_fields = []
for column in model_class.__table__.columns:
column_type = str(column.type)
if 'enum' in column_type.lower() or 'ENUM' in column_type:
# extract possible values of enum
if hasattr(column.type, 'enum_class'):
enum_values = list(column.type.enum_class.__members__.keys())
python_type = "str"
else:
python_type = "str"
else:
python_type = self._sqlalchemy_to_python_type(column.type)
field_info = {
"name": column.name,
"type": column_type,
"python_type": python_type,
"nullable": column.nullable ,
"primary_key": column.primary_key,
"unique": column.unique,
"default": column.default is None,
"is_enum": 'enum' in column_type.lower(),
}
fields.append(field_info)
if not column.nullable and not column.primary_key:
required_fields.append(column.name)
if column.unique:
unique_fields.append(column.name)
return{
"class_name" : model_class.__name__,
"entity_name" : model_class.__name__.lower(),
"model_name" : model_class.__name__,
"model_folder" : model_class.__module__.split(".")[-2] if "." in model_class.__module__ else "models",
"schema_folder" : "schemas" ,
"fields" : fields,
"required_fields" : required_fields,
"unique_fields" : unique_fields,
"has_soft_delete" : any(c.name in ["is_active", "deleted_at"] for c in model_class.__table__.columns),
}
def _sqlalchemy_to_python_type(self, sqlalchemy_type) -> str:
""" Convert SQAlchemy types on a python"""
type_str = str(sqlalchemy_type).lower()
if "int" in type_str:
return "int"
elif "str" in type_str or "varchar" in type_str or "text" in type_str:
return "str"
elif "bool" in type_str:
return "bool"
elif "datetime" in type_str or "date" in type_str:
return "datetime"
elif "float" in type_str or "decimal" in type_str:
return "float"
else:
return "Any"
def generate_service(self, model_metada: Dict[str, Any]):
""" Generate file of service using template"""
template = self.env.get_template("service_template.j2")
output = template.render(**model_metada)
output_file = self.output_dir / "services" /f"{model_metada['entity_name']}_service.py"
with open(output_file, "w", encoding="utf-8") as f:
f.write(output)
print(f"Generated: {output_file}")
def generate_route(self, model_metadata: Dict[str, Any]):
"""Generate file of route using template"""
template = self.env.get_template("route_template.j2")
output = template.render(**model_metadata)
output_file = self.output_dir / "routers" / f"{model_metadata['entity_name']}_router.py"
with open(output_file, "w", encoding="utf-8") as f:
f.write(output)
print(f"Generated: {output_file}")
def generate_all(self, models_list: List):
""" Generate services and router for a list models"""
print(f"Intializing generation")
for model in models_list:
print(f"Processing model: {model.__name__}")
metadata = self.get_model_metadata(model)
if metadata is None:
print(f"Error: can't obtain metadara, jumped \n")
continue
self.generate_service(metadata)
self.generate_route(metadata)
print("Generation completed")
def get_all_models():
"""Import and retunr all models SQLAlchemy"""
from app.modules.coments.models import Coments
from app.modules.configuration.models import Configuration
from app.modules.credits.models import Credits
from app.modules.edos.models import EDOS
from app.modules.efos.models import EFOS
from app.modules.feed.models import Feed
from app.modules.files.models import Files
from app.modules.interactions.models import Interaccion
from app.modules.invoices.models import Invoices
from app.modules.license.models import License
from app.modules.moves.models import Moves
from app.modules.location.models import Locations
from app.modules.suppliers.models import Suppliers
from app.modules.branches.models import Branches
return [Branches, Coments, Configuration, Credits, EDOS, EFOS, Feed, Files, Interaccion, Invoices, License, Moves, Locations, Suppliers]
def main():
#configure routes
generator = CodeGenerator(
templates_dir = "templates",
output_dir = "output/generated"
)
models = get_all_models()
generator.generate_all(models)
if __name__ == "__main__":
main()

5
main.py
View File

@@ -7,6 +7,9 @@ from app.core.servo import get_servo
from database import test_connection
from app.modules.users.route import router as user_router
from app.modules.clients.route import router as client_router
from app.modules.branches.route import router as branches_router
logging.basicConfig(level=logging.INFO,
format= "%(asctime)s - %(levelname)s - %(message)s",
@@ -56,3 +59,5 @@ async def root():
app.include_router(user_router)
app.include_router(client_router)
app.include_router(branches_router)

116
output/generated/routers/branches_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.branches.models import Branches
from app.modules.branches.schema import , Response
from app.modules.branches.services import BranchesService
router = APIRouter(prefix="/branches", tags=["branches"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_branches(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a branches - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create branches")
try:
result = BranchesService.create_branches(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_branchess(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return BranchesService.get_branchess(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_branches_by_id(
branches_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get branches by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = BranchesService.get_branches(db=db, branches_id=branches_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_branches(
branches_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing branches - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update branches")
try:
result = BranchesService.update_branches(
db=db,
branches_id=branches_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_branches(
branches_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete branches by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete branches")
try:
result = BranchesService.delete_branches(db=db, branches_id=branches_id, current_user=current_user)
return {"message": "branches deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

117
output/generated/routers/client_router.py Normal file
View File

@@ -0,0 +1,117 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.client.models import Client
from app.modules.client.schema import , Response
from app.modules.client.services import ClientService
router = APIRouter(prefix="/client", tags=["client"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_client(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a client - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create client")
try:
result = ClientService.create_client(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_clients(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return ClientService.get_clients(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/client_id}", response_model=Response)
def get_client_by_id(
client_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get client by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = ClientService.get_client(db=db, client_id=client_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/client_id}", response_model=Response)
async def update_client(
client_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing client - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update client")
try:
result = ClientService.update_client(
db=db,
client_id=client_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/client_id}", status_code=status.HTTP_200_OK)
async def delete_client(
client_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete client by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete client")
try:
result = ClientService.delete_client(db=db, client_id=client_id, current_user=current_user)
return {"message": "client deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/coments_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.coments.models import Coments
from app.modules.coments.schema import , Response
from app.modules.coments.services import ComentsService
router = APIRouter(prefix="/coments", tags=["coments"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_coments(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a coments - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create coments")
try:
result = ComentsService.create_coments(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_comentss(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return ComentsService.get_comentss(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_coments_by_id(
coments_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get coments by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = ComentsService.get_coments(db=db, coments_id=coments_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_coments(
coments_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing coments - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update coments")
try:
result = ComentsService.update_coments(
db=db,
coments_id=coments_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_coments(
coments_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete coments by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete coments")
try:
result = ComentsService.delete_coments(db=db, coments_id=coments_id, current_user=current_user)
return {"message": "coments deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/configuration_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.configuration.models import Configuration
from app.modules.configuration.schema import , Response
from app.modules.configuration.services import ConfigurationService
router = APIRouter(prefix="/configuration", tags=["configuration"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_configuration(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a configuration - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create configuration")
try:
result = ConfigurationService.create_configuration(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_configurations(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return ConfigurationService.get_configurations(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_configuration_by_id(
configuration_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get configuration by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = ConfigurationService.get_configuration(db=db, configuration_id=configuration_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_configuration(
configuration_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing configuration - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update configuration")
try:
result = ConfigurationService.update_configuration(
db=db,
configuration_id=configuration_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_configuration(
configuration_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete configuration by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete configuration")
try:
result = ConfigurationService.delete_configuration(db=db, configuration_id=configuration_id, current_user=current_user)
return {"message": "configuration deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/credits_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.credits.models import Credits
from app.modules.credits.schema import , Response
from app.modules.credits.services import CreditsService
router = APIRouter(prefix="/credits", tags=["credits"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_credits(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a credits - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create credits")
try:
result = CreditsService.create_credits(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_creditss(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return CreditsService.get_creditss(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_credits_by_id(
credits_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get credits by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = CreditsService.get_credits(db=db, credits_id=credits_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_credits(
credits_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing credits - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update credits")
try:
result = CreditsService.update_credits(
db=db,
credits_id=credits_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_credits(
credits_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete credits by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete credits")
try:
result = CreditsService.delete_credits(db=db, credits_id=credits_id, current_user=current_user)
return {"message": "credits deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/edos_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.edos.models import EDOS
from app.modules.edos.schema import , Response
from app.modules.edos.services import EDOSService
router = APIRouter(prefix="/edos", tags=["edos"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_edos(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a edos - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create edos")
try:
result = EDOSService.create_edos(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_edoss(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return EDOSService.get_edoss(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_edos_by_id(
edos_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get edos by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = EDOSService.get_edos(db=db, edos_id=edos_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_edos(
edos_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing edos - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update edos")
try:
result = EDOSService.update_edos(
db=db,
edos_id=edos_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_edos(
edos_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete edos by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete edos")
try:
result = EDOSService.delete_edos(db=db, edos_id=edos_id, current_user=current_user)
return {"message": "edos deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/efos_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.efos.models import EFOS
from app.modules.efos.schema import , Response
from app.modules.efos.services import EFOSService
router = APIRouter(prefix="/efos", tags=["efos"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_efos(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a efos - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create efos")
try:
result = EFOSService.create_efos(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_efoss(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return EFOSService.get_efoss(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_efos_by_id(
efos_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get efos by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = EFOSService.get_efos(db=db, efos_id=efos_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_efos(
efos_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing efos - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update efos")
try:
result = EFOSService.update_efos(
db=db,
efos_id=efos_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_efos(
efos_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete efos by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete efos")
try:
result = EFOSService.delete_efos(db=db, efos_id=efos_id, current_user=current_user)
return {"message": "efos deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/feed_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.feed.models import Feed
from app.modules.feed.schema import , Response
from app.modules.feed.services import FeedService
router = APIRouter(prefix="/feed", tags=["feed"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_feed(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a feed - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create feed")
try:
result = FeedService.create_feed(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_feeds(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return FeedService.get_feeds(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_feed_by_id(
feed_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get feed by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = FeedService.get_feed(db=db, feed_id=feed_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_feed(
feed_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing feed - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update feed")
try:
result = FeedService.update_feed(
db=db,
feed_id=feed_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_feed(
feed_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete feed by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete feed")
try:
result = FeedService.delete_feed(db=db, feed_id=feed_id, current_user=current_user)
return {"message": "feed deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/files_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.files.models import Files
from app.modules.files.schema import , Response
from app.modules.files.services import FilesService
router = APIRouter(prefix="/files", tags=["files"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_files(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a files - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create files")
try:
result = FilesService.create_files(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_filess(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return FilesService.get_filess(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_files_by_id(
files_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get files by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = FilesService.get_files(db=db, files_id=files_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_files(
files_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing files - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update files")
try:
result = FilesService.update_files(
db=db,
files_id=files_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_files(
files_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete files by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete files")
try:
result = FilesService.delete_files(db=db, files_id=files_id, current_user=current_user)
return {"message": "files deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/invoices_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.invoices.models import Invoices
from app.modules.invoices.schema import , Response
from app.modules.invoices.services import InvoicesService
router = APIRouter(prefix="/invoices", tags=["invoices"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_invoices(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a invoices - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create invoices")
try:
result = InvoicesService.create_invoices(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_invoicess(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return InvoicesService.get_invoicess(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_invoices_by_id(
invoices_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get invoices by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = InvoicesService.get_invoices(db=db, invoices_id=invoices_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_invoices(
invoices_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing invoices - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update invoices")
try:
result = InvoicesService.update_invoices(
db=db,
invoices_id=invoices_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_invoices(
invoices_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete invoices by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete invoices")
try:
result = InvoicesService.delete_invoices(db=db, invoices_id=invoices_id, current_user=current_user)
return {"message": "invoices deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/license_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.license.models import License
from app.modules.license.schema import , Response
from app.modules.license.services import LicenseService
router = APIRouter(prefix="/license", tags=["license"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_license(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a license - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create license")
try:
result = LicenseService.create_license(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_licenses(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return LicenseService.get_licenses(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_license_by_id(
license_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get license by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = LicenseService.get_license(db=db, license_id=license_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_license(
license_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing license - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update license")
try:
result = LicenseService.update_license(
db=db,
license_id=license_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_license(
license_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete license by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete license")
try:
result = LicenseService.delete_license(db=db, license_id=license_id, current_user=current_user)
return {"message": "license deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/locations_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.locations.models import Locations
from app.modules.locations.schema import , Response
from app.modules.locations.services import LocationsService
router = APIRouter(prefix="/locations", tags=["locations"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_locations(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a locations - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create locations")
try:
result = LocationsService.create_locations(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_locationss(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return LocationsService.get_locationss(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_locations_by_id(
locations_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get locations by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = LocationsService.get_locations(db=db, locations_id=locations_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_locations(
locations_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing locations - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update locations")
try:
result = LocationsService.update_locations(
db=db,
locations_id=locations_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_locations(
locations_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete locations by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete locations")
try:
result = LocationsService.delete_locations(db=db, locations_id=locations_id, current_user=current_user)
return {"message": "locations deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/moves_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.moves.models import Moves
from app.modules.moves.schema import , Response
from app.modules.moves.services import MovesService
router = APIRouter(prefix="/moves", tags=["moves"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_moves(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a moves - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create moves")
try:
result = MovesService.create_moves(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_movess(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return MovesService.get_movess(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_moves_by_id(
moves_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get moves by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = MovesService.get_moves(db=db, moves_id=moves_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_moves(
moves_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing moves - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update moves")
try:
result = MovesService.update_moves(
db=db,
moves_id=moves_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_moves(
moves_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete moves by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete moves")
try:
result = MovesService.delete_moves(db=db, moves_id=moves_id, current_user=current_user)
return {"message": "moves deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

116
output/generated/routers/suppliers_router.py Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.models.user import User # Modelo base de usuario
from app.modules.suppliers.models import Suppliers
from app.modules.suppliers.schema import , Response
from app.modules.suppliers.services import SuppliersService
router = APIRouter(prefix="/suppliers", tags=["suppliers"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model=Response, status_code=status.HTTP_201_CREATED)
async def create_suppliers(
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Create a suppliers - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create suppliers")
try:
result = SuppliersService.create_suppliers(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/", response_model=List[Response])
def get_supplierss(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return SuppliersService.get_supplierss(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/{{{entity_name}}_id}", response_model=Response)
def get_suppliers_by_id(
suppliers_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get suppliers by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = SuppliersService.get_suppliers(db=db, suppliers_id=suppliers_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update/{{{entity_name}}_id}", response_model=Response)
async def update_suppliers(
suppliers_id: int,
data: ,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update existing suppliers - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update suppliers")
try:
result = SuppliersService.update_suppliers(
db=db,
suppliers_id=suppliers_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete/{{{entity_name}}_id}", status_code=status.HTTP_200_OK)
async def delete_suppliers(
suppliers_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Delete suppliers by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete suppliers")
try:
result = SuppliersService.delete_suppliers(db=db, suppliers_id=suppliers_id, current_user=current_user)
return {"message": "suppliers deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

67
output/generated/services/branches_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.branches.models import Branches
from app.modules.branches.schema import
class BranchesService:
@staticmethod
def create_branches(db: Session, data: ):
new_branches = Branches(**data.dict())
db.add(new_branches)
db.commit()
db.refresh(new_branches)
return new_branches
@staticmethod
def get_branches(db: Session, branches_id: int, current_user):
branches = db.query(Branches).filter(Branches.id == branches_id).first()
if not branches:
raise ValueError("branches no encontrado")
return branches
@staticmethod
def update_branches(db: Session, branches_id: int, data: , current_user):
branches = db.query(Branches).filter(Branches.id == branches_id).first()
if not branches:
raise ValueError("branches no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este branches")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(branches, key, value)
db.commit()
db.refresh(branches)
return branches
@staticmethod
def delete_branches(db: Session, branches_id: int, current_user):
branches = db.query(Branches).filter(Branches.id == branches_id).first()
if not branches:
raise ValueError("branches no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este branches")
try:
branches.is_active = False
branches.deleted_at = datetime.utcnow()
branches.deleted_by = current_user.id
db.commit()
db.refresh(branches)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el branches: {e}")
return {"message": "branches eliminado correctamente"}

67
output/generated/services/client_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.client.models import Client
from app.modules.client.schema import
class ClientService:
@staticmethod
def create_client(db: Session, data: ):
new_client = Client(**data.dict())
db.add(new_client)
db.commit()
db.refresh(new_client)
return new_client
@staticmethod
def get_client(db: Session, client_id: int, current_user):
client = db.query(Client).filter(Client.id == client_id).first()
if not client:
raise ValueError("client no encontrado")
return client
@staticmethod
def update_client(db: Session, client_id: int, data: , current_user):
client = db.query(Client).filter(Client.id == client_id).first()
if not client:
raise ValueError("client no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este client")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(client, key, value)
db.commit()
db.refresh(client)
return client
@staticmethod
def delete_client(db: Session, client_id: int, current_user):
client = db.query(Client).filter(Client.id == client_id).first()
if not client:
raise ValueError("client no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este client")
try:
client.is_active = False
client.deleted_at = datetime.utcnow()
client.deleted_by = current_user.id
db.commit()
db.refresh(client)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el client: {e}")
return {"message": "client eliminado correctamente"}

67
output/generated/services/coments_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.coments.models import Coments
from app.modules.coments.schema import
class ComentsService:
@staticmethod
def create_coments(db: Session, data: ):
new_coments = Coments(**data.dict())
db.add(new_coments)
db.commit()
db.refresh(new_coments)
return new_coments
@staticmethod
def get_coments(db: Session, coments_id: int, current_user):
coments = db.query(Coments).filter(Coments.id == coments_id).first()
if not coments:
raise ValueError("coments no encontrado")
return coments
@staticmethod
def update_coments(db: Session, coments_id: int, data: , current_user):
coments = db.query(Coments).filter(Coments.id == coments_id).first()
if not coments:
raise ValueError("coments no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este coments")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(coments, key, value)
db.commit()
db.refresh(coments)
return coments
@staticmethod
def delete_coments(db: Session, coments_id: int, current_user):
coments = db.query(Coments).filter(Coments.id == coments_id).first()
if not coments:
raise ValueError("coments no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este coments")
try:
coments.is_active = False
coments.deleted_at = datetime.utcnow()
coments.deleted_by = current_user.id
db.commit()
db.refresh(coments)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el coments: {e}")
return {"message": "coments eliminado correctamente"}

67
output/generated/services/configuration_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.configuration.models import Configuration
from app.modules.configuration.schema import
class ConfigurationService:
@staticmethod
def create_configuration(db: Session, data: ):
new_configuration = Configuration(**data.dict())
db.add(new_configuration)
db.commit()
db.refresh(new_configuration)
return new_configuration
@staticmethod
def get_configuration(db: Session, configuration_id: int, current_user):
configuration = db.query(Configuration).filter(Configuration.id == configuration_id).first()
if not configuration:
raise ValueError("configuration no encontrado")
return configuration
@staticmethod
def update_configuration(db: Session, configuration_id: int, data: , current_user):
configuration = db.query(Configuration).filter(Configuration.id == configuration_id).first()
if not configuration:
raise ValueError("configuration no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este configuration")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(configuration, key, value)
db.commit()
db.refresh(configuration)
return configuration
@staticmethod
def delete_configuration(db: Session, configuration_id: int, current_user):
configuration = db.query(Configuration).filter(Configuration.id == configuration_id).first()
if not configuration:
raise ValueError("configuration no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este configuration")
try:
configuration.is_active = False
configuration.deleted_at = datetime.utcnow()
configuration.deleted_by = current_user.id
db.commit()
db.refresh(configuration)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el configuration: {e}")
return {"message": "configuration eliminado correctamente"}

67
output/generated/services/credits_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.credits.models import Credits
from app.modules.credits.schema import
class CreditsService:
@staticmethod
def create_credits(db: Session, data: ):
new_credits = Credits(**data.dict())
db.add(new_credits)
db.commit()
db.refresh(new_credits)
return new_credits
@staticmethod
def get_credits(db: Session, credits_id: int, current_user):
credits = db.query(Credits).filter(Credits.id == credits_id).first()
if not credits:
raise ValueError("credits no encontrado")
return credits
@staticmethod
def update_credits(db: Session, credits_id: int, data: , current_user):
credits = db.query(Credits).filter(Credits.id == credits_id).first()
if not credits:
raise ValueError("credits no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este credits")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(credits, key, value)
db.commit()
db.refresh(credits)
return credits
@staticmethod
def delete_credits(db: Session, credits_id: int, current_user):
credits = db.query(Credits).filter(Credits.id == credits_id).first()
if not credits:
raise ValueError("credits no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este credits")
try:
credits.is_active = False
credits.deleted_at = datetime.utcnow()
credits.deleted_by = current_user.id
db.commit()
db.refresh(credits)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el credits: {e}")
return {"message": "credits eliminado correctamente"}

67
output/generated/services/edos_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.edos.models import EDOS
from app.modules.edos.schema import
class EDOSService:
@staticmethod
def create_edos(db: Session, data: ):
new_edos = EDOS(**data.dict())
db.add(new_edos)
db.commit()
db.refresh(new_edos)
return new_edos
@staticmethod
def get_edos(db: Session, edos_id: int, current_user):
edos = db.query(EDOS).filter(EDOS.id == edos_id).first()
if not edos:
raise ValueError("edos no encontrado")
return edos
@staticmethod
def update_edos(db: Session, edos_id: int, data: , current_user):
edos = db.query(EDOS).filter(EDOS.id == edos_id).first()
if not edos:
raise ValueError("edos no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este edos")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(edos, key, value)
db.commit()
db.refresh(edos)
return edos
@staticmethod
def delete_edos(db: Session, edos_id: int, current_user):
edos = db.query(EDOS).filter(EDOS.id == edos_id).first()
if not edos:
raise ValueError("edos no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este edos")
try:
edos.is_active = False
edos.deleted_at = datetime.utcnow()
edos.deleted_by = current_user.id
db.commit()
db.refresh(edos)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el edos: {e}")
return {"message": "edos eliminado correctamente"}

67
output/generated/services/efos_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.efos.models import EFOS
from app.modules.efos.schema import
class EFOSService:
@staticmethod
def create_efos(db: Session, data: ):
new_efos = EFOS(**data.dict())
db.add(new_efos)
db.commit()
db.refresh(new_efos)
return new_efos
@staticmethod
def get_efos(db: Session, efos_id: int, current_user):
efos = db.query(EFOS).filter(EFOS.id == efos_id).first()
if not efos:
raise ValueError("efos no encontrado")
return efos
@staticmethod
def update_efos(db: Session, efos_id: int, data: , current_user):
efos = db.query(EFOS).filter(EFOS.id == efos_id).first()
if not efos:
raise ValueError("efos no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este efos")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(efos, key, value)
db.commit()
db.refresh(efos)
return efos
@staticmethod
def delete_efos(db: Session, efos_id: int, current_user):
efos = db.query(EFOS).filter(EFOS.id == efos_id).first()
if not efos:
raise ValueError("efos no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este efos")
try:
efos.is_active = False
efos.deleted_at = datetime.utcnow()
efos.deleted_by = current_user.id
db.commit()
db.refresh(efos)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el efos: {e}")
return {"message": "efos eliminado correctamente"}

67
output/generated/services/feed_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.feed.models import Feed
from app.modules.feed.schema import
class FeedService:
@staticmethod
def create_feed(db: Session, data: ):
new_feed = Feed(**data.dict())
db.add(new_feed)
db.commit()
db.refresh(new_feed)
return new_feed
@staticmethod
def get_feed(db: Session, feed_id: int, current_user):
feed = db.query(Feed).filter(Feed.id == feed_id).first()
if not feed:
raise ValueError("feed no encontrado")
return feed
@staticmethod
def update_feed(db: Session, feed_id: int, data: , current_user):
feed = db.query(Feed).filter(Feed.id == feed_id).first()
if not feed:
raise ValueError("feed no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este feed")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(feed, key, value)
db.commit()
db.refresh(feed)
return feed
@staticmethod
def delete_feed(db: Session, feed_id: int, current_user):
feed = db.query(Feed).filter(Feed.id == feed_id).first()
if not feed:
raise ValueError("feed no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este feed")
try:
feed.is_active = False
feed.deleted_at = datetime.utcnow()
feed.deleted_by = current_user.id
db.commit()
db.refresh(feed)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el feed: {e}")
return {"message": "feed eliminado correctamente"}

67
output/generated/services/files_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.files.models import Files
from app.modules.files.schema import
class FilesService:
@staticmethod
def create_files(db: Session, data: ):
new_files = Files(**data.dict())
db.add(new_files)
db.commit()
db.refresh(new_files)
return new_files
@staticmethod
def get_files(db: Session, files_id: int, current_user):
files = db.query(Files).filter(Files.id == files_id).first()
if not files:
raise ValueError("files no encontrado")
return files
@staticmethod
def update_files(db: Session, files_id: int, data: , current_user):
files = db.query(Files).filter(Files.id == files_id).first()
if not files:
raise ValueError("files no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este files")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(files, key, value)
db.commit()
db.refresh(files)
return files
@staticmethod
def delete_files(db: Session, files_id: int, current_user):
files = db.query(Files).filter(Files.id == files_id).first()
if not files:
raise ValueError("files no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este files")
try:
files.is_active = False
files.deleted_at = datetime.utcnow()
files.deleted_by = current_user.id
db.commit()
db.refresh(files)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el files: {e}")
return {"message": "files eliminado correctamente"}

67
output/generated/services/invoices_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.invoices.models import Invoices
from app.modules.invoices.schema import
class InvoicesService:
@staticmethod
def create_invoices(db: Session, data: ):
new_invoices = Invoices(**data.dict())
db.add(new_invoices)
db.commit()
db.refresh(new_invoices)
return new_invoices
@staticmethod
def get_invoices(db: Session, invoices_id: int, current_user):
invoices = db.query(Invoices).filter(Invoices.id == invoices_id).first()
if not invoices:
raise ValueError("invoices no encontrado")
return invoices
@staticmethod
def update_invoices(db: Session, invoices_id: int, data: , current_user):
invoices = db.query(Invoices).filter(Invoices.id == invoices_id).first()
if not invoices:
raise ValueError("invoices no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este invoices")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(invoices, key, value)
db.commit()
db.refresh(invoices)
return invoices
@staticmethod
def delete_invoices(db: Session, invoices_id: int, current_user):
invoices = db.query(Invoices).filter(Invoices.id == invoices_id).first()
if not invoices:
raise ValueError("invoices no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este invoices")
try:
invoices.is_active = False
invoices.deleted_at = datetime.utcnow()
invoices.deleted_by = current_user.id
db.commit()
db.refresh(invoices)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el invoices: {e}")
return {"message": "invoices eliminado correctamente"}

67
output/generated/services/license_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.license.models import License
from app.modules.license.schema import
class LicenseService:
@staticmethod
def create_license(db: Session, data: ):
new_license = License(**data.dict())
db.add(new_license)
db.commit()
db.refresh(new_license)
return new_license
@staticmethod
def get_license(db: Session, license_id: int, current_user):
license = db.query(License).filter(License.id == license_id).first()
if not license:
raise ValueError("license no encontrado")
return license
@staticmethod
def update_license(db: Session, license_id: int, data: , current_user):
license = db.query(License).filter(License.id == license_id).first()
if not license:
raise ValueError("license no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este license")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(license, key, value)
db.commit()
db.refresh(license)
return license
@staticmethod
def delete_license(db: Session, license_id: int, current_user):
license = db.query(License).filter(License.id == license_id).first()
if not license:
raise ValueError("license no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este license")
try:
license.is_active = False
license.deleted_at = datetime.utcnow()
license.deleted_by = current_user.id
db.commit()
db.refresh(license)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el license: {e}")
return {"message": "license eliminado correctamente"}

67
output/generated/services/locations_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.locations.models import Locations
from app.modules.locations.schema import
class LocationsService:
@staticmethod
def create_locations(db: Session, data: ):
new_locations = Locations(**data.dict())
db.add(new_locations)
db.commit()
db.refresh(new_locations)
return new_locations
@staticmethod
def get_locations(db: Session, locations_id: int, current_user):
locations = db.query(Locations).filter(Locations.id == locations_id).first()
if not locations:
raise ValueError("locations no encontrado")
return locations
@staticmethod
def update_locations(db: Session, locations_id: int, data: , current_user):
locations = db.query(Locations).filter(Locations.id == locations_id).first()
if not locations:
raise ValueError("locations no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este locations")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(locations, key, value)
db.commit()
db.refresh(locations)
return locations
@staticmethod
def delete_locations(db: Session, locations_id: int, current_user):
locations = db.query(Locations).filter(Locations.id == locations_id).first()
if not locations:
raise ValueError("locations no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este locations")
try:
locations.is_active = False
locations.deleted_at = datetime.utcnow()
locations.deleted_by = current_user.id
db.commit()
db.refresh(locations)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el locations: {e}")
return {"message": "locations eliminado correctamente"}

67
output/generated/services/moves_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.moves.models import Moves
from app.modules.moves.schema import
class MovesService:
@staticmethod
def create_moves(db: Session, data: ):
new_moves = Moves(**data.dict())
db.add(new_moves)
db.commit()
db.refresh(new_moves)
return new_moves
@staticmethod
def get_moves(db: Session, moves_id: int, current_user):
moves = db.query(Moves).filter(Moves.id == moves_id).first()
if not moves:
raise ValueError("moves no encontrado")
return moves
@staticmethod
def update_moves(db: Session, moves_id: int, data: , current_user):
moves = db.query(Moves).filter(Moves.id == moves_id).first()
if not moves:
raise ValueError("moves no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este moves")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(moves, key, value)
db.commit()
db.refresh(moves)
return moves
@staticmethod
def delete_moves(db: Session, moves_id: int, current_user):
moves = db.query(Moves).filter(Moves.id == moves_id).first()
if not moves:
raise ValueError("moves no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este moves")
try:
moves.is_active = False
moves.deleted_at = datetime.utcnow()
moves.deleted_by = current_user.id
db.commit()
db.refresh(moves)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el moves: {e}")
return {"message": "moves eliminado correctamente"}

67
output/generated/services/suppliers_service.py Normal file
View File

@@ -0,0 +1,67 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.suppliers.models import Suppliers
from app.modules.suppliers.schema import
class SuppliersService:
@staticmethod
def create_suppliers(db: Session, data: ):
new_suppliers = Suppliers(**data.dict())
db.add(new_suppliers)
db.commit()
db.refresh(new_suppliers)
return new_suppliers
@staticmethod
def get_suppliers(db: Session, suppliers_id: int, current_user):
suppliers = db.query(Suppliers).filter(Suppliers.id == suppliers_id).first()
if not suppliers:
raise ValueError("suppliers no encontrado")
return suppliers
@staticmethod
def update_suppliers(db: Session, suppliers_id: int, data: , current_user):
suppliers = db.query(Suppliers).filter(Suppliers.id == suppliers_id).first()
if not suppliers:
raise ValueError("suppliers no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este suppliers")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr(suppliers, key, value)
db.commit()
db.refresh(suppliers)
return suppliers
@staticmethod
def delete_suppliers(db: Session, suppliers_id: int, current_user):
suppliers = db.query(Suppliers).filter(Suppliers.id == suppliers_id).first()
if not suppliers:
raise ValueError("suppliers no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este suppliers")
try:
suppliers.is_active = False
suppliers.deleted_at = datetime.utcnow()
suppliers.deleted_by = current_user.id
db.commit()
db.refresh(suppliers)
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el suppliers: {e}")
return {"message": "suppliers eliminado correctamente"}

116
templates/route_template.j2 Normal file
View File

@@ -0,0 +1,116 @@
from fastapi import APIRouter, Depends, HTTPException, Query, Header, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from typing import Optional, List
#
from database import get_db
from app.core.auth import get_current_user
from app.modules.users.models import Users # Modelo base de usuario
from app.modules.{{entity_name}}.models import {{model_name}}
from app.modules.{{entity_name}}.schema import {{schema_name}}, {{schema_name}}Response
from app.modules.{{entity_name}}.services import {{class_name}}Service
router = APIRouter(prefix="/{{entity_name}}", tags=["{{entity_name}}"])
security = HTTPBearer()
#================ Create ====================
@router.post("/create", response_model={{schema_name}}Response, status_code=status.HTTP_201_CREATED)
async def create_{{entity_name}}(
data: {{schema_name}},
db: Session = Depends(get_db),
current_user: Users = Depends(get_current_user)
):
"""Create a {{entity_name}} - Requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="User not found - invalid token")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to create {{entity_name}}")
try:
result = {{class_name}}Service.create_{{entity_name}}(db=db, data=data)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
#================ Get ====================
@router.get("/get_list", response_model=List[{{schema_name}}Response])
def get_{{entity_name}}s(
db: Session = Depends(get_db),
current_user: Users = Depends(get_current_user),
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(100, ge=1, le=1000, description="Max number of records to return")
):
"""Get list with pagination - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
return {{class_name}}Service.get_{{entity_name}}s(db=db, skip=skip, limit=limit)
#================ Get by ID ====================
@router.get("/get_Id", response_model={{schema_name}}Response)
def get_{{entity_name}}_by_id(
{{entity_name}}_id: int,
db: Session = Depends(get_db),
current_user: Users = Depends(get_current_user)
):
"""Get {{entity_name}} by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
try:
result = {{class_name}}Service.get_{{entity_name}}(db=db, {{entity_name}}_id={{entity_name}}_id, current_user=current_user)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Update ====================
@router.patch("/update", response_model={{schema_name}}Response)
async def update_{{entity_name}}(
{{entity_name}}_id: int,
data: {{schema_name}},
db: Session = Depends(get_db),
current_user: Users = Depends(get_current_user)
):
"""Update existing {{entity_name}} - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to update {{entity_name}}")
try:
result = {{class_name}}Service.update_{{entity_name}}(
db=db,
{{entity_name}}_id={{entity_name}}_id,
data=data,
current_user=current_user
)
return result
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
#================ Delete ====================
@router.delete("/delete", status_code=status.HTTP_200_OK)
async def delete_{{entity_name}}(
{{entity_name}}_id: int,
db: Session = Depends(get_db),
current_user: Users = Depends(get_current_user)
):
"""Delete {{entity_name}} by ID - requires authentication"""
if not current_user:
raise HTTPException(status_code=401, detail="Authentication required")
if current_user.tipo_usuario not in ["root", "admin"]:
raise HTTPException(status_code=403, detail="You don't have permission to delete {{entity_name}}")
try:
result = {{class_name}}Service.delete_{{entity_name}}(db=db, {{entity_name}}_id={{entity_name}}_id, current_user=current_user)
return {"message": "{{entity_name}} deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))

73
templates/service_template.j2 Normal file
View File

@@ -0,0 +1,73 @@
from sqlalchemy.orm import Session
from datetime import datetime
from typing import Optional
from fastapi import HTTPException
from app.modules.{{entity_name}}.models import {{model_name}}
from app.modules.{{entity_name}}.schema import {{schema_name}}
class {{ class_name }}Service:
@staticmethod
def create_{{ entity_name }}(db: Session, data: {{ schema_name }}):
{% for field in required_fields %}
if not data.{{ field }}:
raise ValueError("{{ field }} es requerido")
{% endfor %}
{% for field in unique_fields %}
if {{ field }}_exists(db, data.{{ field }}):
raise ValueError("{{ field }} ya existe")
{% endfor %}
new_{{ entity_name }} = {{ model_name }}(**data.dict())
db.add(new_{{ entity_name }})
db.commit()
db.refresh(new_{{ entity_name }})
return new_{{ entity_name }}
@staticmethod
def get_{{ entity_name }}(db: Session, {{ entity_name }}_id: int, current_user):
{{ entity_name }} = db.query({{ model_name }}).filter({{ model_name }}.id == {{ entity_name }}_id).first()
if not {{ entity_name }}:
raise ValueError("{{ entity_name }} no encontrado")
return {{ entity_name }}
@staticmethod
def update_{{ entity_name }}(db: Session, {{ entity_name }}_id: int, data: {{ update_schema_name }}, current_user):
{{ entity_name }} = db.query({{ model_name }}).filter({{ model_name }}.id == {{ entity_name }}_id).first()
if not {{ entity_name }}:
raise ValueError("{{ entity_name }} no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para actualizar este {{ entity_name }}")
update_data = data.dict(exclude_unset=True)
for key, value in update_data.items():
setattr({{ entity_name }}, key, value)
db.commit()
db.refresh({{ entity_name }})
return {{ entity_name }}
@staticmethod
def delete_{{ entity_name }}(db: Session, {{ entity_name }}_id: int, current_user):
{{ entity_name }} = db.query({{ model_name }}).filter({{ model_name }}.id == {{ entity_name }}_id).first()
if not {{ entity_name }}:
raise ValueError("{{ entity_name }} no encontrado")
if current_user.role not in ["ROOT", "ADMIN"]:
raise ValueError("No tienes permisos para eliminar este {{ entity_name }}")
try:
{{ entity_name }}.is_active = False
{{ entity_name }}.deleted_at = datetime.utcnow()
{{ entity_name }}.deleted_by = current_user.id
db.commit()
db.refresh({{ entity_name }})
except Exception as e:
db.rollback()
raise ValueError(f"Error al eliminar el {{ entity_name }}: {e}")
return {"message": "{{ entity_name }} eliminado correctamente"}