Feat/account delete

config/dtos/AccountDTO.py

+from pydantic import BaseModel
+from datetime import datetime
+from typing import Optional
+
+
+class AccountCreateDTO(BaseModel):
+    name: str
+    cnpj: str
+
+
+class AccountUpdateDTO(BaseModel):
+    name: Optional[str] = None
+    cnpj: Optional[str] = None
+    is_active: Optional[bool] = None
+
+
+class AccountResponseDTO(BaseModel):
+    id: str
+    name: str
+    cnpj: str
+    is_active: bool
+    created_at: datetime
+    updated_at: datetime
+
+    class Config:
+        from_attributes = True 
+

repository/account_repository.py

 from sqlalchemy.orm import Session
 from model.account import Account
-from sqlalchemy.sql import func
 from datetime import datetime
 
+
+def get_all_accounts(db: Session):
+    return db.query(Account).all()
+
+
 def get_account_by_name(db: Session, name: str):
     return db.query(Account).filter(Account.name == name).all()
 
+
+def get_account_by_id(db: Session, id: int):
+    return db.query(Account).filter(Account.id == id).first()
+
+
 def get_account_by_cnpj(db: Session, cnpj: str):
     return db.query(Account).filter(Account.cnpj == cnpj).first()
 
+
 def get_account_by_is_active(db: Session, is_active: bool):
     return db.query(Account).filter(Account.is_active == is_active).all()
 
-def create_account(db: Session, name: str, cnpj: str):
-    new_account = Account(name=name, cnpj=cnpj, is_active=1, created_at=datetime.utcnow(), updated_at=datetime.utcnow())
+
+def create_account(db: Session, new_account):
+    new_account = Account(
+        name=new_account.get('name'), 
+        cnpj=new_account.get('cnpj'), 
+        is_active=1, 
+        created_at=datetime.now().astimezone(), 
+        updated_at=datetime.now().astimezone()
+    )
     db.add(new_account)
     db.commit()
     db.refresh(new_account)
     return new_account
 
-def get_all_accounts(db: Session):
-    return db.query(Account).all()
\ No newline at end of file
+
+def update_account(db: Session, cnpj, updated_account: Account):
+    account = db.query(Account).filter(Account.cnpj == cnpj).first()
+
+    account.name = updated_account.get('name')
+    account.cnpj = updated_account.get('cnpj')
+    account.is_active = updated_account.get('is_active')
+    account.updated_at = datetime.now().astimezone()
+
+    db.commit()
+    db.refresh(account)
+    return account
+
+
+def delete_account(db: Session, id):
+    account = db.query(Account).filter(Account.id == id).first()
+    account.is_active = 0
+    account.updated_at = datetime.now().astimezone()
+    db.commit()
+    return account
\ No newline at end of file

routers/account_router.py

 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy.orm import Session
 from config.db import SessionLocal
-from service.account_service import get_account_by_name, get_account_by_cnpj, create_new_account, list_accounts, patch_account
+from config.dtos.AccountDTO import AccountCreateDTO, AccountResponseDTO, AccountUpdateDTO
+from service.account_service import get_account, create_new_account, list_accounts, patch_account, delete_account_by_id
 
 router = APIRouter(prefix="/accounts", tags=["Accounts"])
 
@@ -12,43 +13,54 @@ def get_db():
     finally:
         db.close()
 
-@router.get("/{cnpj}")
-def read_account_cnpj(cnpj: str, db: Session = Depends(get_db)):
-    account = get_account_by_cnpj(cnpj, db)
-    if account:
-        return account
-    raise HTTPException(status_code=404, detail="Account not found by CNPJ")
 
-@router.get("/{name}")
-def read_account_name(name: str, db: Session = Depends(get_db)):
-    account = get_account_by_name(name, db)
+@router.get("/")
+def get_account_list(name = None, cnpj = None, db: Session = Depends(get_db)):
+    account = get_account(name, cnpj, db)
     if account:
         return account
-    raise HTTPException(status_code=404, detail="Account not found by name")
+    raise HTTPException(status_code=404, detail="Account not found.")
+
 
 @router.get("/active/{is_active}")
 def get_active_accounts(is_active: bool, db: Session = Depends(get_db)):
     return list_accounts(db)
 
-@router.post("/")
-def add_account(name: str, cnpj: str, db: Session = Depends(get_db)):
+
+@router.post("/", response_model=AccountResponseDTO)
+def add_account(account: AccountCreateDTO, db: Session = Depends(get_db)):
     try:
-        return create_new_account(name, cnpj, db)
+        new_account = { 
+            'name': account.name, 
+            'cnpj': account.cnpj 
+        }
+        return create_new_account(new_account, db)
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
+    
 
-@router.patch("/update")
-def update_account(cnpj: str = None, name: str = None, new_name: str = None, new_cnpj: str = None, db: Session = Depends(get_db)):
+@router.patch("/update", response_model=AccountResponseDTO)
+def update_account(cnpj, account: AccountUpdateDTO, db: Session = Depends(get_db)):
     try:
-        updated_account = patch_account(db, cnpj, name, new_name, new_cnpj)
+        updated_account = { 
+            'cnpj': account.cnpj, 
+            'name': account.name, 
+            'is_active': account.is_active 
+        }
+        updated_account = patch_account(cnpj, updated_account, db)
         return updated_account
     except HTTPException as e:
         raise e 
     except Exception as e:
         raise HTTPException(status_code=400, detail=str(e))
+    
+
+@router.delete("/delete/{id}")
+def delete_account(id, db: Session = Depends(get_db)):
+    try:
+        return delete_account_by_id(id, db)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
 
-@router.get("/")
-def get_accounts(db: Session = Depends(get_db)):
-    return list_accounts(db)
 
 

service/account_service.py

+from fastapi import HTTPException
 from sqlalchemy.orm import Session
 from repository import account_repository
 
-def get_account_by_name(name: str, db: Session):
-    return account_repository.get_account_by_name(db, name)
+def get_account(name: str = None, cnpj: str = None, db: Session = None):
+    if name:
+        return account_repository.get_account_by_name(db, name)
+    if cnpj:
+        return account_repository.get_account_by_cnpj(db, cnpj)  
+    return account_repository.get_all_accounts(db)
 
-def get_account_by_cnpj(cnpj: str, db: Session):
-    return account_repository.get_account_by_cnpj(db, cnpj)
 
-def create_new_account(name: str, cnpj: str, db: Session):
-    existing_account = account_repository.get_account_by_cnpj(db, cnpj)
+def create_new_account(new_account: str, db: Session):
+    existing_account = account_repository.get_account_by_cnpj(db, new_account.get('cnpj'))
     if existing_account:
         raise ValueError("Account with this cnpj already exists.")
     
-    return account_repository.create_account(db, name, cnpj)
+    return account_repository.create_account(db, new_account)
+
 
 def list_accounts(db: Session):
     return account_repository.get_all_accounts(db)
 
+
 def list_active_accounts(db: Session):
     return account_repository.get_account_by_is_active(db)
 
-def patch_account(db: Session, cnpj: str = None, name: str = None, new_name: str = None, new_cnpj: str = None):
-    try:
-        if not cnpj and not name:
-            raise HTTPException(status_code=400, detail="You must provide at least a CNPJ or Name to find the account.")
-        account = None
-
-        if name:
-            account = account_repository.get_account_by_cnpj(db, cnpj)
-        if cnpj:
-            account = account_repository.get_account_by_name(db, name)
 
+def patch_account(cnpj, updated_account, db: Session):
+    try:
+        account = account_repository.get_account_by_cnpj(db, cnpj)
+    
         if not account:
             raise HTTPException(status_code=404, detail="Account not found.")
 
-        if new_name:
-            if account_repository.get_account_by_name(db, new_name) and account.name != new_name:
+        if updated_account.get('name'):
+            if account_repository.get_account_by_name(db, updated_account.get('name')):
                 raise HTTPException(status_code=405, detail="Name is already in use by another account.")
 
-        if new_cnpj:
-            if account_repository.get_account_by_cnpj(db, new_cnpj) and account.cnpj != new_cnpj:
+        if updated_account.get('cnpj'):
+            if account_repository.get_account_by_cnpj(db, cnpj) == updated_account.get('cnpj'):
                 raise HTTPException(status_code=406, detail="CNPJ is already in use by another account.")
 
-        if new_name:
-            account.name = new_name
-        if new_cnpj:
-            account.cnpj = new_cnpj
-
-        db.commit()
-        db.refresh(account)
-        return account
+        return account_repository.update_account(db, cnpj, updated_account)
+    
     except Exception as e:
-        db.rollback()
         raise HTTPException(status_code=400, detail=str(e))
+    
+
+def delete_account_by_id(id, db: Session):
+    account = account_repository.get_account_by_id(db, id)
+    if account:
+        return account_repository.delete_account(db, id)
+    raise HTTPException(status_code=404, detail="Account not found.")
+