52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
from typing import Any, List, Dict
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from personal_info.db.models import Person, Address, Document, Contact
|
|
from personal_info.db.schemas import PersonCreate, PersonBase
|
|
from personal_info.db import models
|
|
from personal_info.services.base import BaseService
|
|
|
|
class PersonService(BaseService[Person, PersonCreate, Any]):
|
|
def __init__(self, db_session: Session):
|
|
super(PersonService, self).__init__(Person, db_session)
|
|
|
|
def list(self, full_name: str, role: str) -> List[Person]:
|
|
if role == "admin" and full_name == "":
|
|
objs: List[Person] = self.db_session.query(Person).all()
|
|
else:
|
|
objs: List[Person] = self.db_session.query(Person).filter(Person.last_name.ilike(full_name)).all()
|
|
return objs
|
|
|
|
def create(self, obj: PersonCreate) -> Person:
|
|
person_obj = PersonBase(**obj.dict())
|
|
person = super(PersonService, self).create(person_obj)
|
|
|
|
addresses = [
|
|
Address(**address.dict(), person_id=person.id) for address in obj.addresses
|
|
]
|
|
self.db_session.add_all(addresses)
|
|
self.db_session.commit()
|
|
|
|
documents = [
|
|
Document(**document.dict(), person_id=person.id) for document in obj.documents
|
|
]
|
|
self.db_session.add_all(documents)
|
|
self.db_session.commit()
|
|
|
|
contacts = [
|
|
Contact(**contact.dict(), person_id=person.id) for contact in obj.contacts
|
|
]
|
|
self.db_session.add_all(contacts)
|
|
self.db_session.commit()
|
|
|
|
return person
|
|
|
|
def batch_get(self, request: Dict) -> List[models.Person]:
|
|
|
|
if request["ids"] is None:
|
|
HTTPException(status_code=404, detail='ids required')
|
|
|
|
objs: List[models.Person] = self.db_session.query(models.Person).filter(models.Person.id.in_(request["ids"])).all()
|
|
|
|
return objs |