diff --git a/backend/tests/test_pycon_registration_usecase.py b/backend/tests/test_pycon_registration_usecase.py new file mode 100644 index 00000000..7615a26d --- /dev/null +++ b/backend/tests/test_pycon_registration_usecase.py @@ -0,0 +1,317 @@ +"""Unit tests for PyconRegistrationUsecase.""" + +from http import HTTPStatus +import unittest +from unittest.mock import MagicMock, patch + +from model.events.events_constants import EventStatus +from model.pycon_registrations.pycon_registration import ( + PyconRegistrationIn, + TicketTypes, +) +from usecase.pycon_registration_usecase import PyconRegistrationUsecase + + +class TestPyconRegistrationUsecase(unittest.TestCase): + """Test suite for PyconRegistrationUsecase.""" + + def setUp(self) -> None: + self.patcher_events = patch( + "usecase.pycon_registration_usecase.EventsRepository" + ) + self.patcher_registrations = patch( + "usecase.pycon_registration_usecase.RegistrationsRepository" + ) + self.patcher_ticket_type = patch( + "usecase.pycon_registration_usecase.TicketTypeRepository" + ) + self.patcher_discounts = patch( + "usecase.pycon_registration_usecase.DiscountUsecase" + ) + self.patcher_payments = patch( + "usecase.pycon_registration_usecase.PaymentTransactionRepository" + ) + self.patcher_email = patch( + "usecase.pycon_registration_usecase.PyConRegistrationEmailNotification" + ) + + self.mock_events_repo = self.patcher_events.start().return_value + self.mock_registrations_repo = self.patcher_registrations.start().return_value + self.mock_ticket_type_repo = self.patcher_ticket_type.start().return_value + self.mock_discount_usecase = self.patcher_discounts.start().return_value + self.mock_payments_repo = self.patcher_payments.start().return_value + self.mock_email_notif = self.patcher_email.start().return_value + + self.usecase = PyconRegistrationUsecase() + + def tearDown(self) -> None: + patch.stopall() + + def _create_mock_event( + self, + paid_event: bool = True, + has_multiple_tickets: bool = True, + sprint_day_price: float = 0.0, + ) -> MagicMock: + event = MagicMock() + event.eventId = "test-event-id" + event.status = EventStatus.OPEN.value + event.registrationCount = 10 + event.isLimitedSlot = False + event.maximumSlots = 100 + event.paidEvent = paid_event + event.hasMultipleTicketTypes = has_multiple_tickets + event.price = 0.0 + event.sprintDayPrice = sprint_day_price + event.maximumSprintDaySlots = 50 + event.sprintDayRegistrationCount = 5 + return event + + def _create_registration_in( + self, + ticket_type: TicketTypes = TicketTypes.CODER, + sprint_day: bool = False, + discount_code: str = "", + transaction_id: str = "", + ) -> PyconRegistrationIn: + return PyconRegistrationIn( + eventId="test-event-id", + email="test@example.com", + firstName="John", + lastName="Doe", + nickname="JD", + pronouns="he/him", + contactNumber="+639123456789", + organization="Tech Corp", + jobTitle="Engineer", + ticketType=ticket_type, + sprintDay=sprint_day, + availTShirt=False, + communityInvolvement=False, + futureVolunteer=False, + discountCode=discount_code, + transactionId=transaction_id, + validIdObjectKey="id-key-123", + ) + + def test_free_ticket_price_succeeds_without_transaction_id(self) -> None: + """When ticket price is 0 on a paidEvent, registration should succeed without a transaction ID.""" + event = self._create_mock_event(paid_event=True, has_multiple_tickets=True) + self.mock_events_repo.query_events.return_value = (HTTPStatus.OK, event, None) + self.mock_registrations_repo.query_registrations_with_email.return_value = ( + HTTPStatus.NOT_FOUND, + [], + None, + ) + + mock_ticket_type = MagicMock() + mock_ticket_type.price = 0.0 + mock_ticket_type.maximumQuantity = 100 + mock_ticket_type.currentSales = 10 + mock_ticket_type.name = "Free Community Pass" + self.mock_ticket_type_repo.query_ticket_type_with_ticket_type_id.return_value = ( + HTTPStatus.OK, + mock_ticket_type, + None, + ) + + mock_stored_reg = MagicMock() + mock_stored_reg.registrationId = "reg-123" + mock_stored_reg.registrationEmailSent = False + mock_stored_reg.email = "test@example.com" + mock_stored_reg.to_simple_dict.return_value = { + "registrationId": "reg-123", + "eventId": "test-event-id", + "email": "test@example.com", + "firstName": "John", + "lastName": "Doe", + "nickname": "JD", + "pronouns": "he/him", + "contactNumber": "+639123456789", + "organization": "Tech Corp", + "jobTitle": "Engineer", + "ticketType": TicketTypes.CODER.value, + "sprintDay": False, + "availTShirt": False, + "communityInvolvement": False, + "futureVolunteer": False, + "validIdObjectKey": "id-key-123", + "createDate": "2026-10-03T18:00:00+08:00", + "updateDate": "2026-10-03T18:00:00+08:00", + } + self.mock_registrations_repo.store_registration.return_value = ( + HTTPStatus.OK, + mock_stored_reg, + None, + ) + self.mock_events_repo.append_event_registration_count.return_value = ( + HTTPStatus.OK, + event, + None, + ) + self.mock_ticket_type_repo.append_ticket_type_sales.return_value = ( + HTTPStatus.OK, + mock_ticket_type, + None, + ) + + reg_in = self._create_registration_in( + ticket_type=TicketTypes.CODER, transaction_id="" + ) + + with patch.object( + self.usecase, "collect_pre_signed_url_pycon", side_effect=lambda x: x + ): + response = self.usecase.create_pycon_registration(reg_in) + + self.mock_payments_repo.query_payment_transaction_with_payment_transaction_id.assert_not_called() + self.mock_registrations_repo.store_registration.assert_called_once() + self.assertEqual(response.email, "test@example.com") + + def test_paid_ticket_price_requires_transaction_id(self) -> None: + """When ticket price is > 0 on a paidEvent, missing transaction ID should return 400 Bad Request.""" + event = self._create_mock_event(paid_event=True, has_multiple_tickets=True) + self.mock_events_repo.query_events.return_value = (HTTPStatus.OK, event, None) + self.mock_registrations_repo.query_registrations_with_email.return_value = ( + HTTPStatus.NOT_FOUND, + [], + None, + ) + + mock_ticket_type = MagicMock() + mock_ticket_type.price = 1500.0 + mock_ticket_type.maximumQuantity = 100 + mock_ticket_type.currentSales = 10 + mock_ticket_type.name = "Professional Ticket" + self.mock_ticket_type_repo.query_ticket_type_with_ticket_type_id.return_value = ( + HTTPStatus.OK, + mock_ticket_type, + None, + ) + + reg_in = self._create_registration_in( + ticket_type=TicketTypes.CODER, transaction_id="" + ) + response = self.usecase.create_pycon_registration(reg_in) + + self.assertEqual(response.status_code, HTTPStatus.BAD_REQUEST) + self.assertIn(b"Transaction ID is required", response.body) + + def test_free_ticket_with_paid_sprint_day_requires_transaction_id(self) -> None: + """When base ticket is 0 but sprint day has a fee, registration must require transaction ID.""" + event = self._create_mock_event( + paid_event=True, has_multiple_tickets=True, sprint_day_price=500.0 + ) + self.mock_events_repo.query_events.return_value = (HTTPStatus.OK, event, None) + self.mock_registrations_repo.query_registrations_with_email.return_value = ( + HTTPStatus.NOT_FOUND, + [], + None, + ) + + mock_ticket_type = MagicMock() + mock_ticket_type.price = 0.0 + mock_ticket_type.maximumQuantity = 100 + mock_ticket_type.currentSales = 10 + mock_ticket_type.name = "Free Pass" + self.mock_ticket_type_repo.query_ticket_type_with_ticket_type_id.return_value = ( + HTTPStatus.OK, + mock_ticket_type, + None, + ) + + reg_in = self._create_registration_in( + ticket_type=TicketTypes.CODER, sprint_day=True, transaction_id="" + ) + response = self.usecase.create_pycon_registration(reg_in) + + self.assertEqual(response.status_code, HTTPStatus.BAD_REQUEST) + self.assertIn(b"Transaction ID is required", response.body) + + def test_paid_ticket_with_100_percent_discount_succeeds_without_transaction_id( + self, + ) -> None: + """When a 100% discount is applied, registration should succeed without a transaction ID.""" + event = self._create_mock_event(paid_event=True, has_multiple_tickets=True) + self.mock_events_repo.query_events.return_value = (HTTPStatus.OK, event, None) + self.mock_registrations_repo.query_registrations_with_email.return_value = ( + HTTPStatus.NOT_FOUND, + [], + None, + ) + + mock_ticket_type = MagicMock() + mock_ticket_type.price = 1500.0 + mock_ticket_type.maximumQuantity = 100 + mock_ticket_type.currentSales = 10 + mock_ticket_type.name = "Professional Ticket" + self.mock_ticket_type_repo.query_ticket_type_with_ticket_type_id.return_value = ( + HTTPStatus.OK, + mock_ticket_type, + None, + ) + + mock_discount = MagicMock() + mock_discount.isReusable = True + mock_discount.remainingUses = 5 + mock_discount.claimed = False + mock_discount.discountPercentage = 1.0 + self.mock_discount_usecase.get_discount.return_value = mock_discount + self.mock_discount_usecase.claim_discount.return_value = None + + mock_stored_reg = MagicMock() + mock_stored_reg.registrationId = "reg-456" + mock_stored_reg.registrationEmailSent = False + mock_stored_reg.email = "discounted@example.com" + mock_stored_reg.to_simple_dict.return_value = { + "registrationId": "reg-456", + "eventId": "test-event-id", + "email": "discounted@example.com", + "firstName": "Jane", + "lastName": "Doe", + "nickname": "JD", + "pronouns": "she/her", + "contactNumber": "+639123456789", + "organization": "Tech Corp", + "jobTitle": "Engineer", + "ticketType": TicketTypes.CODER.value, + "sprintDay": False, + "availTShirt": False, + "communityInvolvement": False, + "futureVolunteer": False, + "discountCode": "DISCOUNT100", + "validIdObjectKey": "id-key-123", + "createDate": "2026-10-03T18:00:00+08:00", + "updateDate": "2026-10-03T18:00:00+08:00", + } + self.mock_registrations_repo.store_registration.return_value = ( + HTTPStatus.OK, + mock_stored_reg, + None, + ) + self.mock_events_repo.append_event_registration_count.return_value = ( + HTTPStatus.OK, + event, + None, + ) + self.mock_ticket_type_repo.append_ticket_type_sales.return_value = ( + HTTPStatus.OK, + mock_ticket_type, + None, + ) + + reg_in = self._create_registration_in( + ticket_type=TicketTypes.CODER, + discount_code="DISCOUNT100", + transaction_id="", + ) + reg_in.email = "discounted@example.com" + + with patch.object( + self.usecase, "collect_pre_signed_url_pycon", side_effect=lambda x: x + ): + response = self.usecase.create_pycon_registration(reg_in) + + self.mock_payments_repo.query_payment_transaction_with_payment_transaction_id.assert_not_called() + self.mock_registrations_repo.store_registration.assert_called_once() + self.assertEqual(response.email, "discounted@example.com") diff --git a/backend/usecase/pycon_registration_usecase.py b/backend/usecase/pycon_registration_usecase.py index 63d7759d..47ad387b 100644 --- a/backend/usecase/pycon_registration_usecase.py +++ b/backend/usecase/pycon_registration_usecase.py @@ -2,9 +2,10 @@ import os import tempfile from http import HTTPStatus -from typing import List, Union +from typing import List, Optional, Tuple, Union import ulid +from model.events.event import Event from model.events.events_constants import EventStatus from model.file_uploads.file_upload import FileDownloadOut from model.pycon_registrations.pycon_registration import ( @@ -12,6 +13,7 @@ PyconRegistrationOut, PyconRegistrationPatch, ) +from model.ticket_types.ticket_types import TicketType from repository.events_repository import EventsRepository from repository.payment_transaction_repository import PaymentTransactionRepository from repository.registrations_repository import RegistrationsRepository @@ -47,6 +49,133 @@ def __init__(self): self.__ticket_type_repository = TicketTypeRepository() self.__payment_transaction_repository = PaymentTransactionRepository() + def __resolve_ticket_type_and_price( + self, event: Event, registration_in: PyconRegistrationIn + ) -> Tuple[Optional[JSONResponse], Optional[TicketType], float]: + """Resolves the ticket type entry and base price for the PyCon registration. + + :param event: The event entity. + :type event: Event + :param registration_in: The incoming registration data. + :type registration_in: PyconRegistrationIn + + :return: A tuple of (error_response, ticket_type_entry, base_price). + :rtype: Tuple[Optional[JSONResponse], Optional[TicketType], float] + """ + if not event.hasMultipleTicketTypes: + return None, None, float(event.price or 0.0) + + ticket_type = getattr(registration_in, "ticketType", None) + ticket_type_id = ticket_type.value if ticket_type else None + if not ticket_type_id: + return ( + JSONResponse( + status_code=HTTPStatus.BAD_REQUEST, + content={ + "message": "Ticket type ID is required for multiple ticket types event" + }, + ), + None, + 0.0, + ) + + status, ticket_type_entry, message = ( + self.__ticket_type_repository.query_ticket_type_with_ticket_type_id( + event_id=event.eventId, ticket_type_id=ticket_type_id + ) + ) + if status != HTTPStatus.OK: + return ( + JSONResponse(status_code=status, content={"message": message}), + None, + 0.0, + ) + + if ticket_type_entry.currentSales >= ticket_type_entry.maximumQuantity: + return ( + JSONResponse( + status_code=HTTPStatus.BAD_REQUEST, + content={ + "message": f"Ticket type {ticket_type_entry.name} is sold out" + }, + ), + None, + 0.0, + ) + + return None, ticket_type_entry, float(ticket_type_entry.price or 0.0) + + def __resolve_discount( + self, event_id: str, discount_code: Optional[str] + ) -> Tuple[Optional[JSONResponse], float]: + """Validates a discount code and resolves its discount percentage. + + :param event_id: The ID of the event. + :type event_id: str + :param discount_code: The discount code entered by the registrant. + :type discount_code: Optional[str] + + :return: A tuple of (error_response, discount_percentage). + :rtype: Tuple[Optional[JSONResponse], float] + """ + if not discount_code: + return None, 0.0 + + discount_entry = self.__discount_usecase.get_discount( + event_id=event_id, entry_id=discount_code + ) + if isinstance(discount_entry, JSONResponse): + return discount_entry, 0.0 + + if discount_entry.isReusable and discount_entry.remainingUses <= 0: + return ( + JSONResponse( + status_code=HTTPStatus.BAD_REQUEST, + content={"message": "Discount code has no remaining uses"}, + ), + 0.0, + ) + + if not discount_entry.isReusable and discount_entry.claimed: + return ( + JSONResponse( + status_code=HTTPStatus.BAD_REQUEST, + content={"message": "Discount code has already been claimed"}, + ), + 0.0, + ) + + return None, float(discount_entry.discountPercentage or 0.0) + + def __validate_payment_transaction( + self, event_id: str, transaction_id: Optional[str] + ) -> Optional[JSONResponse]: + """Validates that a transaction ID is provided and exists for paid events. + + :param event_id: The ID of the event. + :type event_id: str + :param transaction_id: The payment transaction ID. + :type transaction_id: Optional[str] + + :return: An error response if validation fails, or None if valid. + :rtype: Optional[JSONResponse] + """ + if not transaction_id: + return JSONResponse( + status_code=HTTPStatus.BAD_REQUEST, + content={"message": "Transaction ID is required for paid event"}, + ) + + status, _, message = ( + self.__payment_transaction_repository.query_payment_transaction_with_payment_transaction_id( + event_id=event_id, payment_transaction_id=transaction_id + ) + ) + if status != HTTPStatus.OK: + return JSONResponse(status_code=status, content={"message": message}) + + return None + @log_execution def create_pycon_registration( self, registration_in: PyconRegistrationIn @@ -60,17 +189,27 @@ def create_pycon_registration( :rtype: Union[JSONResponse, PyconRegistrationOut] """ - logger.info(f'Saving PyCon registration for email={mask_email(registration_in.email)}, event_id={registration_in.eventId}') - status, event, message = self.__events_repository.query_events(event_id=registration_in.eventId) + logger.info( + f"Saving PyCon registration for email={mask_email(registration_in.email)}, event_id={registration_in.eventId}" + ) + status, event, message = self.__events_repository.query_events( + event_id=registration_in.eventId + ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) event_id = registration_in.eventId - # if registration already exists, return the registration entry - (status, registrations, message) = self.__registrations_repository.query_registrations_with_email(event_id=event_id, email=registration_in.email) + # If registration already exists, return existing registration entry + status, registrations, message = ( + self.__registrations_repository.query_registrations_with_email( + event_id=event_id, email=registration_in.email + ) + ) if status == HTTPStatus.OK and registrations: - logger.info(f'Registration with email {registration_in.email} already exists, returning existing registration') + logger.info( + f"Registration with email {registration_in.email} already exists, returning existing registration" + ) registration = registrations[0] registration_data = self.__convert_data_entry_to_dict(registration) registration_out = PyconRegistrationOut(**registration_data) @@ -80,100 +219,64 @@ def create_pycon_registration( if event.status != EventStatus.OPEN.value: return JSONResponse( status_code=HTTPStatus.BAD_REQUEST, - content={'message': 'Event is not open for registration'}, + content={"message": "Event is not open for registration"}, ) - is_free_ticket = False - - if event.paidEvent: - if registration_in.discountCode: - discount_entry = self.__discount_usecase.get_discount( - event_id=event_id, entry_id=registration_in.discountCode - ) - if isinstance(discount_entry, JSONResponse): - return discount_entry - - if discount_entry.isReusable and discount_entry.remainingUses <= 0: - return JSONResponse( - status_code=HTTPStatus.BAD_REQUEST, content={'message': 'Discount code has no remaining uses'} - ) - - if not discount_entry.isReusable and discount_entry.claimed: - return JSONResponse( - status_code=HTTPStatus.BAD_REQUEST, - content={'message': 'Discount code has already been claimed'}, - ) - - if discount_entry.discountPercentage == 1 and not registration_in.sprintDay: - is_free_ticket = True - - if not is_free_ticket: - transaction_id = registration_in.transactionId - if not transaction_id: - return JSONResponse( - status_code=HTTPStatus.BAD_REQUEST, - content={'message': 'Transaction ID is required for paid event'}, - ) - - ( - status, - _, - message, - ) = self.__payment_transaction_repository.query_payment_transaction_with_payment_transaction_id( - event_id=event_id, payment_transaction_id=transaction_id - ) - if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) - - # Check if the registration with the same email already exists - email = registration_in.email - ( - status, - registrations, - message, - ) = self.__registrations_repository.query_registrations_with_email(event_id=event_id, email=email) - if status == HTTPStatus.OK and registrations: - logger.info(f'Registration with email {email} already exists, returning existing registration') - registration = registrations[0] - registration_data = self.__convert_data_entry_to_dict(registration) - registration_out = PyconRegistrationOut(**registration_data) - return self.collect_pre_signed_url_pycon(registration_out) - - # check if ticket types in event exists + # Check event capacity future_registrations = event.registrationCount if event.isLimitedSlot and future_registrations >= event.maximumSlots: - # check if registration count in event is full return JSONResponse( status_code=HTTPStatus.BAD_REQUEST, - content={'message': f'Event registration is full. Maximum slots: {event.maximumSlots}'}, + content={ + "message": f"Event registration is full. Maximum slots: {event.maximumSlots}" + }, ) + # Check Sprint Day capacity if registration_in.sprintDay: - if event.maximumSprintDaySlots and event.sprintDayRegistrationCount >= event.maximumSprintDaySlots: - return JSONResponse( - status_code=HTTPStatus.BAD_REQUEST, content={'message': 'Sprint Day is already full.'} - ) - - ticket_type_entry = None - if event.hasMultipleTicketTypes: - ticket_type_id = registration_in.ticketType.value - if not ticket_type_id: + if ( + event.maximumSprintDaySlots + and event.sprintDayRegistrationCount >= event.maximumSprintDaySlots + ): return JSONResponse( status_code=HTTPStatus.BAD_REQUEST, - content={'message': 'Ticket type ID is required for multiple ticket types event'}, + content={"message": "Sprint Day is already full."}, ) - status, ticket_type_entry, message = self.__ticket_type_repository.query_ticket_type_with_ticket_type_id( - event_id=event_id, ticket_type_id=ticket_type_id + # Resolve ticket type and base price + error_response, ticket_type_entry, base_price = ( + self.__resolve_ticket_type_and_price(event, registration_in) + ) + if error_response: + return error_response + + # Determine if payment is required + is_free_ticket = False + if not event.paidEvent: + is_free_ticket = True + else: + # Resolve discount percentage if discount code provided + error_response, discount_percentage = self.__resolve_discount( + event_id, registration_in.discountCode ) - if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + if error_response: + return error_response - if ticket_type_entry.currentSales >= ticket_type_entry.maximumQuantity: - return JSONResponse( - status_code=HTTPStatus.BAD_REQUEST, - content={'message': f'Ticket type {ticket_type_entry.name} is sold out'}, + discounted_ticket_price = base_price * (1.0 - discount_percentage) + sprint_day_price = ( + float(event.sprintDayPrice or 0.0) if registration_in.sprintDay else 0.0 + ) + total_amount = discounted_ticket_price + sprint_day_price + + if total_amount <= 0.0: + is_free_ticket = True + + if not is_free_ticket: + error_response = self.__validate_payment_transaction( + event_id, registration_in.transactionId ) + if error_response: + return error_response registration_id = ulid.ulid() discount_code = registration_in.discountCode @@ -186,32 +289,32 @@ def create_pycon_registration( if isinstance(claimed_discount, JSONResponse): return claimed_discount - ( - status, - registration, - message, - ) = self.__registrations_repository.store_registration( - registration_in=registration_in, registration_id=registration_id + status, registration, message = ( + self.__registrations_repository.store_registration( + registration_in=registration_in, registration_id=registration_id + ) ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) logger.info( - f'PyCon registration created successfully: event_id={event_id}, registration_id={registration_id}, email={mask_email(registration_in.email)}' + f"PyCon registration created successfully: event_id={event_id}, registration_id={registration_id}, email={mask_email(registration_in.email)}" ) status, __, message = self.__events_repository.append_event_registration_count( event_entry=event, registration_sprint_day=registration_in.sprintDay ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) if ticket_type_entry: - status, __, message = self.__ticket_type_repository.append_ticket_type_sales( - ticket_type_entry=ticket_type_entry + status, __, message = ( + self.__ticket_type_repository.append_ticket_type_sales( + ticket_type_entry=ticket_type_entry + ) ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) registration_data = self.__convert_data_entry_to_dict(registration) @@ -228,7 +331,10 @@ def create_pycon_registration( @log_execution def update_pycon_registration( - self, event_id: str, registration_id: str, registration_in: PyconRegistrationPatch + self, + event_id: str, + registration_id: str, + registration_in: PyconRegistrationPatch, ) -> Union[JSONResponse, PyconRegistrationOut]: """Updates an existing PyCon registration entry. @@ -247,7 +353,7 @@ def update_pycon_registration( """ status, _, message = self.__events_repository.query_events(event_id=event_id) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) ( status, @@ -257,7 +363,7 @@ def update_pycon_registration( event_id=event_id, registration_id=registration_id ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) ( status, @@ -267,15 +373,19 @@ def update_pycon_registration( registration_entry=registration, registration_in=registration_in ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) - logger.info(f'PyCon registration updated: event_id={event_id}, registration_id={registration_id}') + logger.info( + f"PyCon registration updated: event_id={event_id}, registration_id={registration_id}" + ) registration_data = self.__convert_data_entry_to_dict(update_registration) registration_out = PyconRegistrationOut(**registration_data) return self.collect_pre_signed_url_pycon(registration_out) @log_execution - def get_pycon_registration(self, event_id: str, registration_id: str) -> Union[JSONResponse, PyconRegistrationOut]: + def get_pycon_registration( + self, event_id: str, registration_id: str + ) -> Union[JSONResponse, PyconRegistrationOut]: """Retrieves a specific PyCon registration entry by its ID. :param event_id: The ID of the event @@ -291,7 +401,7 @@ def get_pycon_registration(self, event_id: str, registration_id: str) -> Union[J status, _, message = self.__events_repository.query_events(event_id=event_id) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) ( status, @@ -301,14 +411,16 @@ def get_pycon_registration(self, event_id: str, registration_id: str) -> Union[J event_id=event_id, registration_id=registration_id ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) registration_data = self.__convert_data_entry_to_dict(registration) registration_out = PyconRegistrationOut(**registration_data) return self.collect_pre_signed_url_pycon(registration_out) @log_execution - def get_pycon_registration_by_email(self, event_id: str, email: str) -> Union[JSONResponse, PyconRegistrationOut]: + def get_pycon_registration_by_email( + self, event_id: str, email: str + ) -> Union[JSONResponse, PyconRegistrationOut]: """Retrieves a specific PyCon registration entry by its email. :param event_id: The ID of the event @@ -325,9 +437,11 @@ def get_pycon_registration_by_email(self, event_id: str, email: str) -> Union[JS status, registrations, message, - ) = self.__registrations_repository.query_registrations_with_email(event_id=event_id, email=email) + ) = self.__registrations_repository.query_registrations_with_email( + event_id=event_id, email=email + ) if status != HTTPStatus.OK or not registrations: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) registration = registrations[0] registration_data = self.__convert_data_entry_to_dict(registration) @@ -350,18 +464,22 @@ def get_pycon_registrations( """ status, _, message = self.__events_repository.query_events(event_id=event_id) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) ( status, registrations, message, - ) = self.__registrations_repository.query_registrations(event_id=event_id, is_deleted=is_deleted) + ) = self.__registrations_repository.query_registrations( + event_id=event_id, is_deleted=is_deleted + ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) return [ - self.collect_pre_signed_url_pycon(PyconRegistrationOut(**self.__convert_data_entry_to_dict(registration))) + self.collect_pre_signed_url_pycon( + PyconRegistrationOut(**self.__convert_data_entry_to_dict(registration)) + ) for registration in registrations ] @@ -376,16 +494,18 @@ def get_pycon_registration_csv(self, event_id: str) -> FileDownloadOut: :rtype: FileDownloadOut """ # Get registrations for a PyCon event - status, registrations, message = self.__registrations_repository.query_registrations(event_id=event_id) + status, registrations, message = ( + self.__registrations_repository.query_registrations(event_id=event_id) + ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) try: with tempfile.TemporaryDirectory() as tmpdir: - csv_path = os.path.join(tmpdir, 'pycon_registrations.csv') + csv_path = os.path.join(tmpdir, "pycon_registrations.csv") - with open(csv_path, 'w') as temp: + with open(csv_path, "w") as temp: writer = csv.writer(temp) # make the first row csv for the keys using the dict keys of the first entry @@ -398,17 +518,21 @@ def get_pycon_registration_csv(self, event_id: str) -> FileDownloadOut: writer.writerow(entry_dict.values()) # upload the file to s3 - csv_object_key = f'csv/pycon_registrations/{event_id}.csv' - self.__file_s3_usecase.upload_file(file_name=csv_path, object_name=csv_object_key) + csv_object_key = f"csv/pycon_registrations/{event_id}.csv" + self.__file_s3_usecase.upload_file( + file_name=csv_path, object_name=csv_object_key + ) return self.__file_s3_usecase.create_download_url(csv_object_key) except Exception as e: - logger.error(f'Error generating the PyCon CSV for {event_id}: {e}') + logger.error(f"Error generating the PyCon CSV for {event_id}: {e}") return @log_execution - def delete_pycon_registration(self, event_id: str, registration_id: str) -> Union[None, JSONResponse]: + def delete_pycon_registration( + self, event_id: str, registration_id: str + ) -> Union[None, JSONResponse]: """Deletes a specific PyCon registration entry by its ID. :param event_id: The ID of the event @@ -423,7 +547,7 @@ def delete_pycon_registration(self, event_id: str, registration_id: str) -> Unio """ status, _, message = self.__events_repository.query_events(event_id=event_id) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) ( status, @@ -433,17 +557,23 @@ def delete_pycon_registration(self, event_id: str, registration_id: str) -> Unio event_id=event_id, registration_id=registration_id ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) - status, message = self.__registrations_repository.delete_registration(registration_entry=registration) + status, message = self.__registrations_repository.delete_registration( + registration_entry=registration + ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) - logger.info(f'PyCon registration deleted: event_id={event_id}, registration_id={registration_id}') + logger.info( + f"PyCon registration deleted: event_id={event_id}, registration_id={registration_id}" + ) return None @log_execution - def collect_pre_signed_url_pycon(self, registration: PyconRegistrationOut) -> PyconRegistrationOut: + def collect_pre_signed_url_pycon( + self, registration: PyconRegistrationOut + ) -> PyconRegistrationOut: """Collects the pre-signed URL for the valid ID image for PyCon registrations. :param registration: The PyCon registration entry to be updated. @@ -454,7 +584,9 @@ def collect_pre_signed_url_pycon(self, registration: PyconRegistrationOut) -> Py """ if registration.validIdObjectKey: - image_id_url = self.__file_s3_usecase.create_download_url(registration.validIdObjectKey) + image_id_url = self.__file_s3_usecase.create_download_url( + registration.validIdObjectKey + ) registration.imageIdUrl = image_id_url.downloadLink return registration @@ -473,26 +605,35 @@ def resend_confirmation_email(self, event_id: str, email: str): :rtype: JSONResponse """ - status, event, message = self.__events_repository.query_events(event_id=event_id) + status, event, message = self.__events_repository.query_events( + event_id=event_id + ) if status != HTTPStatus.OK: - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) - (status, registrations, message) = self.__registrations_repository.query_registrations_with_email( - event_id=event_id, email=email + (status, registrations, message) = ( + self.__registrations_repository.query_registrations_with_email( + event_id=event_id, email=email + ) ) if status == HTTPStatus.OK and registrations and registrations[0].transactionId: registration = registrations[0] - logger.info(f'Resending confirmation email for event {event_id} to {mask_email(email)}') + logger.info( + f"Resending confirmation email for event {event_id} to {mask_email(email)}" + ) self.__pycon_email_notification.send_registration_success_email( email=email, event=event, is_pycon_event=True, registration_data=registration, ) - return JSONResponse(status_code=HTTPStatus.OK, content={'message': f'Confirmation email sent to {email}'}) + return JSONResponse( + status_code=HTTPStatus.OK, + content={"message": f"Confirmation email sent to {email}"}, + ) - return JSONResponse(status_code=status, content={'message': message}) + return JSONResponse(status_code=status, content={"message": message}) @staticmethod def __convert_data_entry_to_dict(data_entry):