Coverage for src/kwai/modules/club/repositories/flemish_member_importer.py: 77%

1"""Module for defining an importer for members of the Flemish Judo Federation.""" 

2 

3import csv 

4 

5from typing import Any, AsyncGenerator 

6 

7from kwai.core.domain.value_objects.date import Date 

8from kwai.core.domain.value_objects.email_address import ( 

9 EmailAddress, 

10 InvalidEmailException, 

11) 

12from kwai.core.domain.value_objects.name import Name 

13from kwai.core.domain.value_objects.owner import Owner 

14from kwai.modules.club.domain.contact import ContactEntity 

15from kwai.modules.club.domain.member import MemberEntity 

16from kwai.modules.club.domain.person import PersonEntity 

17from kwai.modules.club.domain.value_objects import Address, Birthdate, Gender, License 

18from kwai.modules.club.repositories.country_repository import ( 

19 CountryNotFoundException, 

20 CountryRepository, 

21) 

22from kwai.modules.club.repositories.member_importer import ( 

23 FailureResult, 

24 MemberImporter, 

25 OkResult, 

26 Result, 

27) 

28 

29 

30class FlemishMemberImporter(MemberImporter): 

31 """A class for importing members of the Flemish Judo Federation. 

32 

33 The import is a csv file. 

34 """ 

35 

36 def __init__(self, filename: str, owner: Owner, country_repo: CountryRepository): 

37 """Initialize the importer. 

38 

39 Args: 

40 filename: The name of the csv file. 

41 owner: The user that started the upload. 

42 country_repo: A repository to get the nationality of a member. 

43 """ 

44 super().__init__(filename, owner, country_repo) 

45 

46 async def import_(self) -> AsyncGenerator[Result, None]: 

47 with open(self._filename) as csv_file: 

48 member_reader = csv.DictReader(csv_file) 

49 row: dict[str, Any] 

50 for row_index, row in enumerate(member_reader): 

51 if row["geslacht"] == "V": 

52 gender = Gender.FEMALE 

53 elif row["geslacht"] == "M": 

54 gender = Gender.MALE 

55 else: 

56 gender = Gender.UNKNOWN 

57 

58 try: 

59 nationality = await self._get_country( 

60 self._country_repo, row["nationaliteit"] 

61 ) 

62 except CountryNotFoundException: 

63 yield FailureResult( 

64 row=row_index, 

65 message=f"Unrecognized country: {row['nationaliteit']}", 

66 ) 

67 continue 

68 

69 emails = [] 

70 try: 

71 for email in row["email"].split(";"): 

72 emails.append(EmailAddress(email.strip())) 

73 except InvalidEmailException as exc: 

74 yield FailureResult(row=row_index, message=str(exc)) 

75 continue 

76 

77 try: 

78 country = await self._get_country(self._country_repo, row["land"]) 

79 except CountryNotFoundException: 

80 yield FailureResult( 

81 row=row_index, 

82 message=f"Unrecognized country: {row['land']}", 

83 ) 

84 continue 

85 

86 yield OkResult( 

87 row=row_index, 

88 member=MemberEntity( 

89 license=License( 

90 number=row["vergunning"], 

91 end_date=Date.create_from_string(row["vervaldatum"]), 

92 ), 

93 person=PersonEntity( 

94 name=Name( 

95 first_name=row["voornaam"], last_name=row["achternaam"] 

96 ), 

97 gender=gender, 

98 birthdate=Birthdate( 

99 Date.create_from_string(row["geboortedatum"]) 

100 ), 

101 nationality=nationality, 

102 contact=ContactEntity( 

103 emails=emails, 

104 address=Address( 

105 address=row["straatnummer"], 

106 postal_code=row["postnummer"], 

107 city=row["gemeente"], 

108 county="", 

109 country=country, 

110 ), 

111 mobile=row["telefoon1"], 

112 tel=row["telefoon2"], 

113 ), 

114 ), 

115 active=row["status"] == "ACTIEF", 

116 ), 

117 ) 

118 self._get_country.cache_clear()