Coverage for bc/kwai-bc-club/src/kwai_bc_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 

14 

15from kwai_bc_club.domain.contact import ContactEntity 

16from kwai_bc_club.domain.member import MemberEntity 

17from kwai_bc_club.domain.person import PersonEntity 

18from kwai_bc_club.domain.value_objects import Address, Birthdate, Gender, License 

19from kwai_bc_club.repositories.country_repository import ( 

20 CountryNotFoundException, 

21 CountryRepository, 

22) 

23from kwai_bc_club.repositories.member_importer import ( 

24 FailureResult, 

25 MemberImporter, 

26 OkResult, 

27 Result, 

28) 

29 

30 

31class FlemishMemberImporter(MemberImporter): 

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

33 

34 The import is a csv file. 

35 """ 

36 

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

38 """Initialize the importer. 

39 

40 Args: 

41 filename: The name of the csv file. 

42 owner: The user that started the upload. 

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

44 """ 

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

46 

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

48 with open(self._filename) as csv_file: 

49 member_reader = csv.DictReader(csv_file) 

50 row: dict[str, Any] 

51 for row_index, row in enumerate(member_reader): 

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

53 gender = Gender.FEMALE 

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

55 gender = Gender.MALE 

56 else: 

57 gender = Gender.UNKNOWN 

58 

59 try: 

60 nationality = await self._get_country( 

61 self._country_repo, row["nationaliteit"] 

62 ) 

63 except CountryNotFoundException: 

64 yield FailureResult( 

65 row=row_index, 

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

67 ) 

68 continue 

69 

70 emails = [] 

71 try: 

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

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

74 except InvalidEmailException as exc: 

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

76 continue 

77 

78 try: 

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

80 except CountryNotFoundException: 

81 yield FailureResult( 

82 row=row_index, 

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

84 ) 

85 continue 

86 

87 yield OkResult( 

88 row=row_index, 

89 member=MemberEntity( 

90 license=License( 

91 number=row["vergunning"], 

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

93 ), 

94 person=PersonEntity( 

95 name=Name( 

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

97 ), 

98 gender=gender, 

99 birthdate=Birthdate( 

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

101 ), 

102 nationality=nationality, 

103 contact=ContactEntity( 

104 emails=tuple(emails), 

105 address=Address( 

106 address=row["straatnummer"], 

107 postal_code=row["postnummer"], 

108 city=row["gemeente"], 

109 county="", 

110 country=country, 

111 ), 

112 mobile=row["telefoon1"], 

113 tel=row["telefoon2"], 

114 ), 

115 ), 

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

117 ), 

118 ) 

119 self._get_country.cache_clear()