diff --git a/README.md b/README.md new file mode 100644 index 0000000..e79493d --- /dev/null +++ b/README.md @@ -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. diff --git a/__pycache__/database.cpython-311.pyc b/__pycache__/database.cpython-311.pyc new file mode 100644 index 0000000..cc5f5bc Binary files /dev/null and b/__pycache__/database.cpython-311.pyc differ diff --git a/app/core/__pycache__/dependencies.cpython-311.pyc b/app/core/__pycache__/dependencies.cpython-311.pyc new file mode 100644 index 0000000..e76c20c Binary files /dev/null and b/app/core/__pycache__/dependencies.cpython-311.pyc differ diff --git a/app/core/dependencies.py b/app/core/dependencies.py index e69de29..f705878 100644 --- a/app/core/dependencies.py +++ b/app/core/dependencies.py @@ -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 + \ No newline at end of file diff --git a/app/modules/affidavit_logs/schema.py b/app/modules/affidavit_logs/schema.py index e69de29..fdb0fdb 100644 --- a/app/modules/affidavit_logs/schema.py +++ b/app/modules/affidavit_logs/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/branches/__pycache__/route.cpython-311.pyc b/app/modules/branches/__pycache__/route.cpython-311.pyc new file mode 100644 index 0000000..964708a Binary files /dev/null and b/app/modules/branches/__pycache__/route.cpython-311.pyc differ diff --git a/app/modules/branches/__pycache__/schema.cpython-311.pyc b/app/modules/branches/__pycache__/schema.cpython-311.pyc new file mode 100644 index 0000000..3fc3dd2 Binary files /dev/null and b/app/modules/branches/__pycache__/schema.cpython-311.pyc differ diff --git a/app/modules/branches/__pycache__/service.cpython-311.pyc b/app/modules/branches/__pycache__/service.cpython-311.pyc new file mode 100644 index 0000000..63b377c Binary files /dev/null and b/app/modules/branches/__pycache__/service.cpython-311.pyc differ diff --git a/app/modules/branches/route.py b/app/modules/branches/route.py index e69de29..54a375e 100644 --- a/app/modules/branches/route.py +++ b/app/modules/branches/route.py @@ -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)) \ No newline at end of file diff --git a/app/modules/branches/schema.py b/app/modules/branches/schema.py index e69de29..fe7cf4c 100644 --- a/app/modules/branches/schema.py +++ b/app/modules/branches/schema.py @@ -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 + diff --git a/app/modules/branches/service.py b/app/modules/branches/service.py new file mode 100644 index 0000000..a814b73 --- /dev/null +++ b/app/modules/branches/service.py @@ -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"} + + \ No newline at end of file diff --git a/app/modules/clients/__pycache__/models.cpython-311.pyc b/app/modules/clients/__pycache__/models.cpython-311.pyc index 3087ae7..fafa4e4 100644 Binary files a/app/modules/clients/__pycache__/models.cpython-311.pyc and b/app/modules/clients/__pycache__/models.cpython-311.pyc differ diff --git a/app/modules/clients/__pycache__/repository.cpython-311.pyc b/app/modules/clients/__pycache__/repository.cpython-311.pyc new file mode 100644 index 0000000..d5cfa4c Binary files /dev/null and b/app/modules/clients/__pycache__/repository.cpython-311.pyc differ diff --git a/app/modules/clients/__pycache__/route.cpython-311.pyc b/app/modules/clients/__pycache__/route.cpython-311.pyc new file mode 100644 index 0000000..140da70 Binary files /dev/null and b/app/modules/clients/__pycache__/route.cpython-311.pyc differ diff --git a/app/modules/clients/__pycache__/schema.cpython-311.pyc b/app/modules/clients/__pycache__/schema.cpython-311.pyc new file mode 100644 index 0000000..0e4502d Binary files /dev/null and b/app/modules/clients/__pycache__/schema.cpython-311.pyc differ diff --git a/app/modules/clients/__pycache__/service.cpython-311.pyc b/app/modules/clients/__pycache__/service.cpython-311.pyc new file mode 100644 index 0000000..b0950ef Binary files /dev/null and b/app/modules/clients/__pycache__/service.cpython-311.pyc differ diff --git a/app/modules/clients/models.py b/app/modules/clients/models.py index 467c9a9..0285867 100644 --- a/app/modules/clients/models.py +++ b/app/modules/clients/models.py @@ -74,3 +74,4 @@ class Client(Base): return f"" + \ No newline at end of file diff --git a/app/modules/clients/repository.py b/app/modules/clients/repository.py index e69de29..526f672 100644 --- a/app/modules/clients/repository.py +++ b/app/modules/clients/repository.py @@ -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) diff --git a/app/modules/clients/route.py b/app/modules/clients/route.py index e69de29..3fd0f33 100644 --- a/app/modules/clients/route.py +++ b/app/modules/clients/route.py @@ -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"]) diff --git a/app/modules/clients/schema.py b/app/modules/clients/schema.py index e69de29..8213182 100644 --- a/app/modules/clients/schema.py +++ b/app/modules/clients/schema.py @@ -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 diff --git a/app/modules/clients/service.py b/app/modules/clients/service.py new file mode 100644 index 0000000..9a9c70f --- /dev/null +++ b/app/modules/clients/service.py @@ -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 + diff --git a/app/modules/coments/schema.py b/app/modules/coments/schema.py index e69de29..27448b1 100644 --- a/app/modules/coments/schema.py +++ b/app/modules/coments/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/configuration/schema.py b/app/modules/configuration/schema.py index e69de29..8400551 100644 --- a/app/modules/configuration/schema.py +++ b/app/modules/configuration/schema.py @@ -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 + \ No newline at end of file diff --git a/app/modules/credits/schema.py b/app/modules/credits/schema.py index e69de29..efc50a4 100644 --- a/app/modules/credits/schema.py +++ b/app/modules/credits/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/edos/schema.py b/app/modules/edos/schema.py index e69de29..efc50a4 100644 --- a/app/modules/edos/schema.py +++ b/app/modules/edos/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/efos/schema.py b/app/modules/efos/schema.py index e69de29..efc50a4 100644 --- a/app/modules/efos/schema.py +++ b/app/modules/efos/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/entreprise/schema.py b/app/modules/entreprise/schema.py index e69de29..efc50a4 100644 --- a/app/modules/entreprise/schema.py +++ b/app/modules/entreprise/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/feed/schema.py b/app/modules/feed/schema.py index e69de29..efc50a4 100644 --- a/app/modules/feed/schema.py +++ b/app/modules/feed/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/files/schema.py b/app/modules/files/schema.py index e69de29..efc50a4 100644 --- a/app/modules/files/schema.py +++ b/app/modules/files/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/interactions/schema.py b/app/modules/interactions/schema.py index e69de29..efc50a4 100644 --- a/app/modules/interactions/schema.py +++ b/app/modules/interactions/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/invoices/schema.py b/app/modules/invoices/schema.py index e69de29..efc50a4 100644 --- a/app/modules/invoices/schema.py +++ b/app/modules/invoices/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/license/schema.py b/app/modules/license/schema.py index e69de29..efc50a4 100644 --- a/app/modules/license/schema.py +++ b/app/modules/license/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/location/service.py b/app/modules/location/service.py index e69de29..efc50a4 100644 --- a/app/modules/location/service.py +++ b/app/modules/location/service.py @@ -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 \ No newline at end of file diff --git a/app/modules/moves/schema.py b/app/modules/moves/schema.py index e69de29..efc50a4 100644 --- a/app/modules/moves/schema.py +++ b/app/modules/moves/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/reports/schema.py b/app/modules/reports/schema.py index e69de29..efc50a4 100644 --- a/app/modules/reports/schema.py +++ b/app/modules/reports/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/suppliers/service.py b/app/modules/suppliers/service.py index e69de29..efc50a4 100644 --- a/app/modules/suppliers/service.py +++ b/app/modules/suppliers/service.py @@ -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 \ No newline at end of file diff --git a/app/modules/taxes/schema.py b/app/modules/taxes/schema.py index e69de29..efc50a4 100644 --- a/app/modules/taxes/schema.py +++ b/app/modules/taxes/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/taxpayers/schema.py b/app/modules/taxpayers/schema.py index e69de29..efc50a4 100644 --- a/app/modules/taxpayers/schema.py +++ b/app/modules/taxpayers/schema.py @@ -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 \ No newline at end of file diff --git a/app/modules/users/__pycache__/route.cpython-311.pyc b/app/modules/users/__pycache__/route.cpython-311.pyc index b1615dd..3e6e422 100644 Binary files a/app/modules/users/__pycache__/route.cpython-311.pyc and b/app/modules/users/__pycache__/route.cpython-311.pyc differ diff --git a/app/pipelines/__pycache__/base.cpython-311.pyc b/app/pipelines/__pycache__/base.cpython-311.pyc index 577faa8..557c98e 100644 Binary files a/app/pipelines/__pycache__/base.cpython-311.pyc and b/app/pipelines/__pycache__/base.cpython-311.pyc differ diff --git a/app/pipelines/__pycache__/crud.cpython-311.pyc b/app/pipelines/__pycache__/crud.cpython-311.pyc index 2420b27..fd93d8f 100644 Binary files a/app/pipelines/__pycache__/crud.cpython-311.pyc and b/app/pipelines/__pycache__/crud.cpython-311.pyc differ diff --git a/app/pipelines/base.py b/app/pipelines/base.py index 7bb8b41..42fc4b0 100644 --- a/app/pipelines/base.py +++ b/app/pipelines/base.py @@ -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 diff --git a/app/pipelines/crud.py b/app/pipelines/crud.py index 2d4d889..ae52a80 100644 --- a/app/pipelines/crud.py +++ b/app/pipelines/crud.py @@ -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()}") diff --git a/generator.py b/generator.py new file mode 100644 index 0000000..6bffc60 --- /dev/null +++ b/generator.py @@ -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() + diff --git a/main.py b/main.py index 88ea359..d39f554 100644 --- a/main.py +++ b/main.py @@ -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) \ No newline at end of file diff --git a/output/generated/routers/branches_router.py b/output/generated/routers/branches_router.py new file mode 100644 index 0000000..03b041f --- /dev/null +++ b/output/generated/routers/branches_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/client_router.py b/output/generated/routers/client_router.py new file mode 100644 index 0000000..d8969e1 --- /dev/null +++ b/output/generated/routers/client_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/coments_router.py b/output/generated/routers/coments_router.py new file mode 100644 index 0000000..5a1756f --- /dev/null +++ b/output/generated/routers/coments_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/configuration_router.py b/output/generated/routers/configuration_router.py new file mode 100644 index 0000000..b3f9980 --- /dev/null +++ b/output/generated/routers/configuration_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/credits_router.py b/output/generated/routers/credits_router.py new file mode 100644 index 0000000..b448a00 --- /dev/null +++ b/output/generated/routers/credits_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/edos_router.py b/output/generated/routers/edos_router.py new file mode 100644 index 0000000..1e140af --- /dev/null +++ b/output/generated/routers/edos_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/efos_router.py b/output/generated/routers/efos_router.py new file mode 100644 index 0000000..20543e0 --- /dev/null +++ b/output/generated/routers/efos_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/feed_router.py b/output/generated/routers/feed_router.py new file mode 100644 index 0000000..9adf2fe --- /dev/null +++ b/output/generated/routers/feed_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/files_router.py b/output/generated/routers/files_router.py new file mode 100644 index 0000000..5c9db99 --- /dev/null +++ b/output/generated/routers/files_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/invoices_router.py b/output/generated/routers/invoices_router.py new file mode 100644 index 0000000..91a341d --- /dev/null +++ b/output/generated/routers/invoices_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/license_router.py b/output/generated/routers/license_router.py new file mode 100644 index 0000000..7e090a9 --- /dev/null +++ b/output/generated/routers/license_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/locations_router.py b/output/generated/routers/locations_router.py new file mode 100644 index 0000000..cf5d4fd --- /dev/null +++ b/output/generated/routers/locations_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/moves_router.py b/output/generated/routers/moves_router.py new file mode 100644 index 0000000..abc5100 --- /dev/null +++ b/output/generated/routers/moves_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/routers/suppliers_router.py b/output/generated/routers/suppliers_router.py new file mode 100644 index 0000000..5ce9cf6 --- /dev/null +++ b/output/generated/routers/suppliers_router.py @@ -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)) \ No newline at end of file diff --git a/output/generated/services/branches_service.py b/output/generated/services/branches_service.py new file mode 100644 index 0000000..8bd8794 --- /dev/null +++ b/output/generated/services/branches_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/client_service.py b/output/generated/services/client_service.py new file mode 100644 index 0000000..37c2c3f --- /dev/null +++ b/output/generated/services/client_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/coments_service.py b/output/generated/services/coments_service.py new file mode 100644 index 0000000..a5dd49c --- /dev/null +++ b/output/generated/services/coments_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/configuration_service.py b/output/generated/services/configuration_service.py new file mode 100644 index 0000000..90b6ff7 --- /dev/null +++ b/output/generated/services/configuration_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/credits_service.py b/output/generated/services/credits_service.py new file mode 100644 index 0000000..5a2deb1 --- /dev/null +++ b/output/generated/services/credits_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/edos_service.py b/output/generated/services/edos_service.py new file mode 100644 index 0000000..a13a5d1 --- /dev/null +++ b/output/generated/services/edos_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/efos_service.py b/output/generated/services/efos_service.py new file mode 100644 index 0000000..2a1a488 --- /dev/null +++ b/output/generated/services/efos_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/feed_service.py b/output/generated/services/feed_service.py new file mode 100644 index 0000000..8af60aa --- /dev/null +++ b/output/generated/services/feed_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/files_service.py b/output/generated/services/files_service.py new file mode 100644 index 0000000..eb588d5 --- /dev/null +++ b/output/generated/services/files_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/invoices_service.py b/output/generated/services/invoices_service.py new file mode 100644 index 0000000..295d046 --- /dev/null +++ b/output/generated/services/invoices_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/license_service.py b/output/generated/services/license_service.py new file mode 100644 index 0000000..ba082c5 --- /dev/null +++ b/output/generated/services/license_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/locations_service.py b/output/generated/services/locations_service.py new file mode 100644 index 0000000..58319a6 --- /dev/null +++ b/output/generated/services/locations_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/moves_service.py b/output/generated/services/moves_service.py new file mode 100644 index 0000000..23e0357 --- /dev/null +++ b/output/generated/services/moves_service.py @@ -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"} + + \ No newline at end of file diff --git a/output/generated/services/suppliers_service.py b/output/generated/services/suppliers_service.py new file mode 100644 index 0000000..0d81e79 --- /dev/null +++ b/output/generated/services/suppliers_service.py @@ -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"} + + \ No newline at end of file diff --git a/templates/route_template.j2 b/templates/route_template.j2 new file mode 100644 index 0000000..955ac9d --- /dev/null +++ b/templates/route_template.j2 @@ -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)) \ No newline at end of file diff --git a/templates/service_template.j2 b/templates/service_template.j2 new file mode 100644 index 0000000..a3ee7b9 --- /dev/null +++ b/templates/service_template.j2 @@ -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"} + + \ No newline at end of file