Coverage for src / avisos / services / aviso_service.py: 100%

1"""Regras de negocio do modulo avisos.""" 

2 

3from sqlalchemy import select 

4from sqlalchemy.orm import Session 

5 

6from src.avisos.models import Aviso 

7from src.avisos.schemas import AvisoCreateSchema, AvisoUpdateSchema 

8 

9 

10def list_avisos(db: Session) -> list[Aviso]: 

11 """Lista avisos cadastrados.""" 

12 return list(db.scalars(select(Aviso).order_by(Aviso.id))) 

13 

14 

15def get_aviso_by_id(db: Session, aviso_id: int) -> Aviso | None: 

16 """Busca aviso por id.""" 

17 return db.get(Aviso, aviso_id) 

18 

19 

20def create_aviso(db: Session, payload: AvisoCreateSchema) -> Aviso: 

21 """Cria aviso e persiste no banco.""" 

22 aviso = Aviso(**payload.model_dump()) 

23 db.add(aviso) 

24 db.commit() 

25 db.refresh(aviso) 

26 return aviso 

27 

28 

29def update_aviso( 

30 db: Session, aviso: Aviso, payload: AvisoCreateSchema 

31) -> Aviso: 

32 """Substitui os dados do aviso existente (PUT).""" 

33 for key, value in payload.model_dump().items(): 

34 setattr(aviso, key, value) 

35 

36 db.add(aviso) 

37 db.commit() 

38 db.refresh(aviso) 

39 return aviso 

40 

41 

42def patch_aviso( 

43 db: Session, aviso: Aviso, payload: AvisoUpdateSchema 

44) -> Aviso: 

45 """Atualiza parcialmente os dados do aviso (PATCH).""" 

46 for key, value in payload.model_dump(exclude_unset=True).items(): 

47 setattr(aviso, key, value) 

48 

49 db.add(aviso) 

50 db.commit() 

51 db.refresh(aviso) 

52 return aviso 

53 

54 

55def delete_aviso(db: Session, aviso: Aviso) -> None: 

56 """Remove aviso do banco.""" 

57 db.delete(aviso) 

58 db.commit()