Coverage for src/kwai/modules/club/repositories/person_db_repository.py: 100%

1"""Module that implements a person repository for a database.""" 

2 

3from dataclasses import dataclass 

4 

5from sql_smith.functions import alias, on 

6 

7from kwai.core.db.database import Database 

8from kwai.core.db.table_row import JoinedTableRow 

9from kwai.core.domain.entity import Entity 

10from kwai.modules.club.domain.person import PersonEntity, PersonIdentifier 

11from kwai.modules.club.repositories._tables import ( 

12 ContactRow, 

13 CountryRow, 

14 PersonRow, 

15) 

16from kwai.modules.club.repositories.contact_db_repository import ContactDbRepository 

17from kwai.modules.club.repositories.person_repository import ( 

18 PersonNotFoundException, 

19 PersonRepository, 

20) 

21 

22 

23@dataclass(kw_only=True, frozen=True, slots=True) 

24class PersonQueryRow(JoinedTableRow): 

25 """A data transfer object for a Contact query.""" 

26 

27 person: PersonRow 

28 contact: ContactRow 

29 country: CountryRow 

30 nationality: CountryRow 

31 

32 def create_entity(self) -> PersonEntity: 

33 """Create a Contact entity from a row.""" 

34 return self.person.create_entity( 

35 self.nationality.create_country(), 

36 self.contact.create_entity(self.country.create_country()), 

37 ) 

38 

39 

40class PersonDbRepository(PersonRepository): 

41 """A person repository for a database.""" 

42 

43 def __init__(self, database: Database): 

44 self._database = database 

45 

46 async def create(self, person: PersonEntity) -> PersonEntity: 

47 if person.contact.id.is_empty(): 

48 new_contact = await ContactDbRepository(self._database).create( 

49 person.contact 

50 ) 

51 person = Entity.replace(person, contact=new_contact) 

52 new_id = await self._database.insert( 

53 PersonRow.__table_name__, PersonRow.persist(person) 

54 ) 

55 return Entity.replace(person, id_=PersonIdentifier(new_id)) 

56 

57 async def update(self, person: PersonEntity) -> None: 

58 await ContactDbRepository(self._database).update(person.contact) 

59 await self._database.update( 

60 person.id.value, PersonRow.__table_name__, PersonRow.persist(person) 

61 ) 

62 

63 async def delete(self, person: PersonEntity): 

64 await ContactDbRepository(self._database).delete(person.contact) 

65 await self._database.delete(person.id.value, PersonRow.__table_name__) 

66 

67 async def get(self, id_: PersonIdentifier) -> PersonEntity: 

68 query = Database.create_query_factory().select() 

69 query.from_(PersonRow.__table_name__).columns( 

70 *PersonQueryRow.get_aliases() 

71 ).inner_join( 

72 ContactRow.__table_name__, 

73 on(ContactRow.column("id"), PersonRow.column("contact_id")), 

74 ).inner_join( 

75 CountryRow.__table_name__, 

76 on(CountryRow.column("id"), ContactRow.column("country_id")), 

77 ).inner_join( 

78 alias(CountryRow.__table_name__, "nationality"), 

79 on("nationality.id", PersonRow.column("nationality_id")), 

80 ).where(PersonRow.field("id").eq(id_.value)) 

81 

82 row = await self._database.fetch_one(query) 

83 if row: 

84 return PersonQueryRow.map(row).create_entity() 

85 

86 raise PersonNotFoundException(f"Person with {id_} not found")