Api reference

components.global_profile.public.api

Address dataclass

Address(
    street,
    postal_code,
    locality,
    country,
    unit=None,
    administrative_area=None,
    sublocality=None,
    postal_town=None,
)

Represents a physical address with international support.

Attributes:

Name	Type	Description
street	str	The street name and number (e.g., "123 Main Street")
unit	str | None	Additional address details like apartment number, suite, or floor (e.g., "Apt 4B", "Suite 200")
locality	str | None	The city or town name (e.g., "Paris", "London")
administrative_area	str | None	The state, province, region, or department (e.g., "Île-de-France", "California")
postal_code	str | None	The ZIP code or postal code (e.g., "75001", "90210")
country	str	The ISO 3166-1 alpha-2 for the country.
sublocality	str | None	A smaller division within a locality such as district, borough, or neighborhood (e.g., "Le Marais", "Brooklyn")
postal_town	str | None	The post town or postal locality, which may differ from the locality (particularly relevant in some countries like the UK)

Example

address = Address(
    street="123 Main Street",
    unit="Apt 4B",
    locality="Paris",
    administrative_area="Île-de-France",
    postal_code="75001",
    country="FR",
    sublocality="Le Marais",
    postal_town=None
)

Errors

ValueError: for invalid country code or invalid administrative area code AddressInvalidException: for country specific validation e.g. Canada location and postal codes

__post_init__

__post_init__()

Enforce ISO 3166 code validation

Source code in components/global_profile/internal/domain/entities.py

def __post_init__(self) -> None:
    """
    Enforce ISO 3166 code validation
    """

    # Sanitize inputs
    sanitized_values = {
        "street": self.street.strip(),
        "country": self.country.strip(),
        "unit": self.unit.strip() if self.unit is not None else None,
        "locality": self.locality.strip() if self.locality is not None else None,
        "administrative_area": (
            self.administrative_area.strip()
            if self.administrative_area is not None
            else None
        ),
        "postal_code": (
            self.postal_code.strip() if self.postal_code is not None else None
        ),
        "sublocality": (
            self.sublocality.strip() if self.sublocality is not None else None
        ),
        "postal_town": (
            self.postal_town.strip() if self.postal_town is not None else None
        ),
    }

    # Country specific sanitize inputs
    if sanitized_values["country"] == "CA" and self.postal_code is not None:
        sanitized_values["postal_code"] = self.postal_code.replace(" ", "").upper()

    # Update instance with sanitized values
    if any(getattr(self, key) != value for key, value in sanitized_values.items()):
        for key, value in sanitized_values.items():
            object.__setattr__(self, key, value)

    # Validate country code
    if pycountry.countries.get(alpha_2=self.country) is None:
        raise ValueError(
            f"Invalid country code: {self.country}; it must be ISO 3166-1 alpha-2!"
        )
    # Validate administrative area if present
    if self.administrative_area:
        if pycountry.subdivisions.get(code=self.administrative_area) is None:  # type: ignore[no-untyped-call]
            raise ValueError(
                f"Invalid administrative area code: {self.administrative_area}; it must be ISO 3166-2!"
            )
    # Validate existence of Canadian addresses
    if self.country == "CA":
        if self._postal_code_ca_re.match(mandatory(self.postal_code)) is None:
            raise AddressInvalidException()
    elif self.country == "FR":
        if not self.locality:
            raise ValueError("Locality cannot be empty for 🇫🇷 addresses")

        if not self._postal_code_fr_re.match(mandatory(self.postal_code)):
            raise ValueError(
                f"{self.postal_code} does not match the expected postal code for 🇫🇷"
            )

administrative_area class-attribute instance-attribute

administrative_area = None

country instance-attribute

country

locality instance-attribute

locality

postal_code instance-attribute

postal_code

postal_town class-attribute instance-attribute

postal_town = None

street instance-attribute

street

sublocality class-attribute instance-attribute

sublocality = None

unit class-attribute instance-attribute

unit = None

ProfileService

ProfileService(unit_of_work=None, message_bus=None)

Entry point to interact with profiles

This service exposes use cases, it is not intended as a CRUD service. As a result, when updating this service you should make sure that: - You introduce a new use case that is not already covered by the existing methods - Your change do not impact subscribers to current events, meaning that they should not be interested by your new event and won't miss any data change they are interested in

Note: Adding events should be carefully considered as granularity of such events can be quite complex to get right.

Instantiates the ProfileService

Default values are provided for target production use, you may need to override them.

Source code in components/global_profile/public/api.py

def __init__(
    self,
    unit_of_work: UnitOfWork | None = None,
    message_bus: MessageBus | None = None,
):
    """
    Instantiates the ProfileService

    Default values are provided for target production use, you may need to override them.

    """
    self.unit_of_work: UnitOfWork = unit_of_work or TransactionUnitOfWork()

    # handlers (domainevent + command) are expected to accept two arguments:
    # * the event they want to handle
    # * and an argument named unit_of_work being the relevant unit_of_work
    # here we take a handler(e, unit_of_work) and turn into a handler(e)
    injected_event_handlers = {
        event_type: [
            # technically that cast is incorrect since handle don't accept any random commands
            cast(
                "Callable[[DomainEvent], None]",
                functools.partial(handler, unit_of_work=self.unit_of_work),
            )
            for handler in handlers
        ]
        for event_type, handlers in _EVENT_HANDLERS.items()
    }
    injected_command_handlers = {
        command_type: [
            # technically that cast is incorrect since handle don't accept any random commands
            cast(
                "Callable[[Command], None]",
                functools.partial(handler, unit_of_work=self.unit_of_work),
            )
            for handler in handlers
        ]
        for command_type, handlers in _COMMAND_HANDLERS.items()
    }

    self.message_bus: MessageBus = message_bus or MessageBus(
        self.unit_of_work, injected_event_handlers, injected_command_handlers
    )

    # User Compatibility module
    self.user_compat: UserCompatibility = GlobalUserCompatibility(self.unit_of_work)

change_address

change_address(profile_id, *, address)

Change the address of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def change_address(
    self,
    profile_id: uuid.UUID,
    *,
    address: Address,
) -> None:
    """Change the address of a profile"""
    command = ChangeAddressCommand(profile_id, address)
    with self.unit_of_work:
        self.message_bus.handle(command)

change_birthdate

change_birthdate(profile_id, *, birthdate)

Change the birthdate of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def change_birthdate(
    self,
    profile_id: uuid.UUID,
    *,
    birthdate: date,
) -> None:
    """Change the birthdate of a profile"""
    command = ChangeBirthdateCommand(
        profile_id=profile_id,
        birthdate=birthdate,
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

change_contact_information

change_contact_information(profile_id, *, email, address)

Change the contact information (address, email) of a profile Doesn't change the value if None is provided. To delete, please check the specific delete methods.

Source code in components/global_profile/public/api.py

@obs.api_call()
def change_contact_information(
    self,
    profile_id: uuid.UUID,
    *,
    email: str,
    address: Address | None,
) -> None:
    """Change the contact information (address, email) of a profile
    Doesn't change the value if None is provided. To delete, please check the specific delete methods.
    """
    command = ChangeContactInformationCommand(
        profile_id,
        email,
        address,
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

change_email

change_email(profile_id, *, email)

Change the email address of a profile

Source code in components/global_profile/public/api.py

def change_email(self, profile_id: uuid.UUID, *, email: str) -> None:
    """Change the email address of a profile"""
    command = ChangeEmailCommand(profile_id, email)

    with self.unit_of_work:
        self.message_bus.handle(command)

change_first_name

change_first_name(profile_id, *, first_name)

Change the first name of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def change_first_name(
    self,
    profile_id: uuid.UUID,
    *,
    first_name: str,
) -> None:
    """Change the first name of a profile"""
    command = ChangeFirstNameCommand(
        profile_id=profile_id,
        first_name=first_name,
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

change_gender

change_gender(profile_id, *, gender)

Change the gender of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def change_gender(
    self,
    profile_id: uuid.UUID,
    *,
    gender: Gender,
) -> None:
    """Change the gender of a profile"""
    command = ChangeGenderCommand(
        profile_id=profile_id,
        gender=GenderTransformer.to_domain(gender),
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

change_identity_information

change_identity_information(
    profile_id,
    *,
    first_name=None,
    last_name=None,
    birth_date=None
)

THIS METHOD IS FOR MIGRATION PURPOSE ONLY, DO NOT USE IT FOR OTHER PURPOSE

Change the identity information (first name, last name, birthdate) of a profile

⚠️ None means the information will be deleted. To use carefully

Source code in components/global_profile/public/api.py

@obs.api_call()
@deprecated("For migration purpose only", category=AlanDeprecationWarning)
def change_identity_information(
    self,
    profile_id: uuid.UUID,
    *,
    first_name: str | None = None,
    last_name: str | None = None,
    birth_date: date | None = None,
) -> None:
    """
    THIS METHOD IS FOR MIGRATION PURPOSE ONLY, DO NOT USE IT FOR OTHER PURPOSE

    Change the identity information (first name, last name, birthdate) of a profile

    ⚠️ `None` means the information will be deleted. To use carefully
    """
    command = ChangeIdentityInformationCommand(
        profile_id,
        first_name,
        last_name,
        birth_date,
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

change_last_name

change_last_name(profile_id, *, last_name)

Change the last name of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def change_last_name(
    self,
    profile_id: uuid.UUID,
    *,
    last_name: str,
) -> None:
    """Change the last name of a profile"""
    command = ChangeLastNameCommand(
        profile_id=profile_id,
        last_name=last_name,
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

change_phone_number

change_phone_number(profile_id, *, phone_number)

Change the phone number of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def change_phone_number(
    self, profile_id: uuid.UUID, *, phone_number: str | None
) -> None:
    """Change the phone number of a profile"""
    command = ChangePhoneNumberCommand(profile_id, phone_number)
    with self.unit_of_work:
        self.message_bus.handle(command)

change_preferred_language

change_preferred_language(
    profile_id, *, preferred_language
)

Change the preferred language of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def change_preferred_language(
    self,
    profile_id: uuid.UUID,
    *,
    preferred_language: Language,
) -> None:
    """Change the preferred language of a profile"""
    command = ChangePreferredLanguageCommand(
        profile_id, LanguageTransformer.to_domain(preferred_language)
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

complete_information

complete_information(
    profile_id,
    *,
    first_name,
    last_name,
    birth_date,
    allow_override=False
)

Complete the information of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def complete_information(
    self,
    profile_id: uuid.UUID,
    *,
    first_name: str,
    last_name: str,
    birth_date: date | None,
    allow_override: bool = False,
) -> None:
    """Complete the information of a profile"""
    command = CompleteInformationCommand(
        profile_id, first_name, last_name, birth_date, allow_override=allow_override
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

create classmethod

create(app_name=None)

Creates a ProfileService instance with the default parameters for a given country: Currently implemented countries: ES, FR, BE

With global_instance, it will use a specific repository allowing cross-country access. It adds profile to the "main" repository given the current environment, update all repositories with the given profile id and reads the first value found given a lookup table.

Source code in components/global_profile/public/api.py

@classmethod
def create(cls, app_name: AppName | None = None) -> "ProfileService":
    """
    Creates a ProfileService instance with the default parameters for a given country:
    Currently implemented countries: ES, FR, BE

    With global_instance, it will use a specific repository allowing cross-country access.
    It adds profile to the "main" repository given the current environment, update all repositories with the given profile id and reads the first value found given a lookup table.
    """
    from components.global_profile.internal.infrastructure.unit_of_work import (
        TransactionUnitOfWork,
    )

    app_name = app_name or get_current_app_name()

    if app_name == AppName.ALAN_CA:
        # TODO: @crew-global-architecture remove the ca user repo once the shared packages are migrated to use profile only
        from components.ca.internal.user.profile.repository import (  # noqa:ALN039, ALN043
            CaUserProfileRepository,
        )

        return cls(
            unit_of_work=TransactionUnitOfWork(
                profile_repository_factory=CaUserProfileRepository
            )
        )
    elif app_name == AppName.ALAN_ES:
        # ⚠️ only load GlobalUserProfileRepository on the app that load all models from all components
        from components.global_profile.internal.infrastructure.global_user_repository import (
            GlobalUserProfileRepository,
        )

        return cls(
            unit_of_work=TransactionUnitOfWork(
                profile_repository_factory=lambda session: GlobalUserProfileRepository(
                    session=session, app_name=app_name
                ),
            )
        )
    elif app_name == AppName.ALAN_BE:
        # ⚠️ only load GlobalUserProfileRepository on the app that load all models from all components
        from components.global_profile.internal.infrastructure.global_user_repository import (
            GlobalUserProfileRepository,
        )

        return cls(
            unit_of_work=TransactionUnitOfWork(
                profile_repository_factory=lambda session: GlobalUserProfileRepository(
                    session=session, app_name=app_name
                ),
            )
        )
    elif app_name == AppName.SHARED_TESTING:
        return cls.create_testing()
    else:
        # ⚠️ only load GlobalUserProfileRepository on the app that load all models from all components
        from components.global_profile.internal.infrastructure.global_user_repository import (
            GlobalUserProfileRepository,
        )

        return cls(
            unit_of_work=TransactionUnitOfWork(
                profile_repository_factory=lambda session: GlobalUserProfileRepository(
                    session=session, app_name=app_name
                )
            )
        )

create_profile

create_profile(
    *,
    profile_id=None,
    email=None,
    first_name=None,
    last_name=None,
    additional_first_names=None,
    birth_date=None,
    birth_name=None,
    gender=None,
    preferred_language=Language.ENGLISH
)

Create a new profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def create_profile(
    self,
    *,
    profile_id: uuid.UUID | None = None,
    email: str | None = None,
    first_name: str | None = None,
    last_name: str | None = None,
    additional_first_names: list[str] | None = None,
    birth_date: date | None = None,
    birth_name: str | None = None,
    gender: Gender | None = None,
    preferred_language: Language = Language.ENGLISH,
) -> uuid.UUID:
    """Create a new profile"""
    selected_profile_id = profile_id if profile_id else uuid.uuid4()

    command = CreateProfileCommand(
        profile_id=selected_profile_id,
        email=email,
        first_name=first_name,
        last_name=last_name,
        additional_first_names=additional_first_names,
        birth_date=birth_date,
        birth_name=birth_name,
        gender=GenderTransformer.to_domain(gender),
        preferred_language=LanguageTransformer.to_domain(preferred_language),
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

    return selected_profile_id

create_testing classmethod

create_testing()

Creates a ProfileService instance for the shared TestUser

Source code in components/global_profile/public/api.py

@classmethod
def create_testing(cls) -> "ProfileService":
    """
    Creates a ProfileService instance for the shared TestUser
    """
    from components.global_profile.public.tests.test_user_repository import (
        TestUserProfileRepositoryV2,
    )

    return cls(
        unit_of_work=TransactionUnitOfWork(
            profile_repository_factory=TestUserProfileRepositoryV2
        )
    )

delete_address

delete_address(profile_id)

Delete the address of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def delete_address(self, profile_id: uuid.UUID) -> None:
    """
    Delete the address of a profile
    """
    command = DeleteAddressCommand(profile_id)

    with self.unit_of_work:
        self.message_bus.handle(command)

delete_email

delete_email(profile_id, *, redacted_value=None)

Delete the email address of a profile

Source code in components/global_profile/public/api.py

def delete_email(
    self, profile_id: uuid.UUID, *, redacted_value: str | None = None
) -> None:
    """Delete the email address of a profile"""
    command = DeleteEmailCommand(
        profile_id=profile_id, redacted_value=redacted_value
    )

    with self.unit_of_work:
        self.message_bus.handle(command)

delete_gender

delete_gender(profile_id)

Delete the gender of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def delete_gender(self, profile_id: uuid.UUID) -> None:
    """
    Delete the gender of a profile
    """
    command = DeleteGenderCommand(profile_id)

    with self.unit_of_work:
        self.message_bus.handle(command)

delete_phone_number

delete_phone_number(profile_id)

Delete the phone number of a profile

Source code in components/global_profile/public/api.py

@obs.api_call()
def delete_phone_number(self, profile_id: uuid.UUID) -> None:
    """
    Delete the phone number of a profile
    """
    command = DeletePhoneNumberCommand(profile_id)

    with self.unit_of_work:
        self.message_bus.handle(command)

edit_birth_information

edit_birth_information(
    profile_id,
    *,
    birth_date=None,
    birth_name=None,
    place_of_birth=None
)

Updates the birth information of a profile after a manual edition (e.g. from the member, the admin...)

None means the information won't be updated: values cannot be unset using this method.

Source code in components/global_profile/public/api.py

@obs.api_call()
def edit_birth_information(
    self,
    profile_id: uuid.UUID,
    *,
    birth_date: date | None = None,
    birth_name: str | None = None,
    place_of_birth: PlaceOfBirth | None = None,
) -> None:
    """
    Updates the birth information of a profile after a manual edition (e.g. from the member, the admin...)

    `None` means the information won't be updated: values cannot be unset using this method.
    """
    command = EditBirthInformationCommand(
        profile_id,
        birth_date,
        birth_name,
        place_of_birth,
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

edit_consents

edit_consents(profile_id, *, old_consents, new_consents)

Updates the consents of a profile after a manual edition from the member

Source code in components/global_profile/public/api.py

@obs.api_call()
def edit_consents(
    self,
    profile_id: uuid.UUID,
    *,
    old_consents: ProfileConsents | None,
    new_consents: EditConsentPayload,
) -> None:
    """
    Updates the consents of a profile after a manual edition from the member
    """
    updated_at = datetime.now(UTC).isoformat()

    # TODO: Consider a specialized advisory lock as concurrent updates could potentially be possible
    # (eg. if multiple consent updates are triggered by fast clicks on several toggles)

    consents = old_consents or ProfileConsents()

    for new_consent in new_consents.values:
        old_consent = next(
            (c for c in consents.values if c.type == new_consent.type), None
        )
        if not old_consent or old_consent.value != new_consent.value:
            consents = replace(
                consents,
                values=[c for c in consents.values if c.type != new_consent.type]
                + [
                    ProfileConsent(
                        type=new_consent.type,
                        value=new_consent.value,
                        updated_at=updated_at,
                    )
                ],
            )

    command = EditConsentsCommand(profile_id, consents=consents)
    with self.unit_of_work:
        self.message_bus.handle(command)

edit_personal_information

edit_personal_information(
    profile_id,
    *,
    first_name=None,
    last_name=None,
    additional_first_names=None,
    birth_date=None
)

Updates the personal information of a profile after a manual edition (e.g. from the member, the admin...)

None means the information won't be updated: values cannot be unset using this method.

birth_date is deprecated and will be removed in a future release. Please use edit_birth_information instead.

Source code in components/global_profile/public/api.py

@obs.api_call()
def edit_personal_information(
    self,
    profile_id: uuid.UUID,
    *,
    first_name: str | None = None,
    last_name: str | None = None,
    additional_first_names: list[str] | None = None,
    birth_date: date | None = None,  # Deprecated: do not use, will be removed
) -> None:
    """
    Updates the personal information of a profile after a manual edition (e.g. from the member, the admin...)

    `None` means the information won't be updated: values cannot be unset using this method.

    `birth_date` is deprecated and will be removed in a future release.
    Please use `edit_birth_information` instead.
    """
    command = EditPersonalInformationCommand(
        profile_id,
        first_name,
        last_name,
        additional_first_names,
        birth_date,
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

fill_missing_information

fill_missing_information(
    profile_id,
    *,
    first_name=None,
    last_name=None,
    additional_first_names=None,
    birth_name=None,
    birth_date=None,
    place_of_birth=None,
    gender=None
)

Fills missing information of a Profile.

It won't replace any existing values, only fills the one set to None.

Source code in components/global_profile/public/api.py

@obs.api_call()
def fill_missing_information(
    self,
    profile_id: uuid.UUID,
    *,
    first_name: str | None = None,
    last_name: str | None = None,
    additional_first_names: list[str] | None = None,
    birth_name: str | None = None,
    birth_date: date | None = None,
    place_of_birth: PlaceOfBirth | None = None,
    gender: Gender | None = None,
) -> None:
    """
    Fills missing information of a Profile.

    It won't replace any existing values, only fills the one set to None.
    """
    command = FillMissingInformationCommand(
        profile_id=profile_id,
        first_name=first_name,
        last_name=last_name,
        additional_first_names=additional_first_names,
        birth_name=birth_name,
        birth_date=birth_date,
        place_of_birth=place_of_birth,
        gender=gender,
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

get_or_raise_profile

get_or_raise_profile(profile_id)

Get a profile by its ID or raise an error if not found

Source code in components/global_profile/public/api.py

@obs.api_call()
def get_or_raise_profile(self, profile_id: uuid.UUID) -> Profile:
    """Get a profile by its ID or raise an error if not found"""
    with self.unit_of_work:
        profile = self.unit_of_work.profiles.get(profile_id)
    if profile is None:
        raise BaseErrorCode.missing_resource(
            message=f"Profile {profile_id} not found"
        )
    return profile

get_or_raise_profiles

get_or_raise_profiles(profile_ids)

Get multiple profiles by their IDs or raise an error if not found

Source code in components/global_profile/public/api.py

@obs.api_call()
def get_or_raise_profiles(
    self, profile_ids: Iterable[uuid.UUID | str]
) -> list[Profile]:
    """Get multiple profiles by their IDs or raise an error if not found"""
    with self.unit_of_work:
        profiles = self.unit_of_work.profiles.get_many(
            [uuid.UUID(str(profile_id)) for profile_id in profile_ids]
        )
    if len(profiles) != len(set(profile_ids)):
        missing_ids = set(profile_ids) - {profile.id for profile in profiles}
        raise BaseErrorCode.missing_resource(
            message=f"Profiles {missing_ids} not found"
        )
    return cast("list[Profile]", profiles)

get_profile

get_profile(profile_id)

Get a profile by its ID

Source code in components/global_profile/public/api.py

@obs.api_call()
def get_profile(self, profile_id: uuid.UUID) -> Profile | None:
    """Get a profile by its ID"""
    with self.unit_of_work:
        return self.unit_of_work.profiles.get(profile_id)

get_profile_by_email

get_profile_by_email(email)

Get a profile by its email address

Source code in components/global_profile/public/api.py

@obs.api_call()
def get_profile_by_email(self, email: str) -> Profile | None:
    """Get a profile by its email address"""
    with self.unit_of_work:
        return self.unit_of_work.profiles.get_by_email(email)

get_profiles

get_profiles(profile_ids)

Get multiple profiles by their IDs

Source code in components/global_profile/public/api.py

@obs.api_call()
def get_profiles(self, profile_ids: Iterable[uuid.UUID | str]) -> list[Profile]:
    """Get multiple profiles by their IDs"""
    with self.unit_of_work:
        profiles = self.unit_of_work.profiles.get_many(
            [uuid.UUID(str(profile_id)) for profile_id in profile_ids]
        )
        return cast("list[Profile]", profiles)

merge_profile_into_another

merge_profile_into_another(
    *,
    source_profile_id,
    target_profile_id,
    strategy=ProfileMergeStrategy.SOURCE_DELETED
)

Merges one profile into another. Merging profile A (source) to B (target) will simply delete profile A and emit an event ProfileMergedEvent with the IDs of both profiles for models to be updated accordingly.

Source code in components/global_profile/public/api.py

@obs.api_call()
def merge_profile_into_another(
    self,
    *,
    source_profile_id: uuid.UUID,
    target_profile_id: uuid.UUID,
    strategy: ProfileMergeStrategy = ProfileMergeStrategy.SOURCE_DELETED,
) -> None:
    """
    Merges one profile into another.
    Merging profile A (source) to B (target) will simply delete profile A and emit an event `ProfileMergedEvent` with the IDs of both profiles for models to be updated accordingly.
    """
    command = MergeProfilesCommand(
        source_profile_id=source_profile_id,
        target_profile_id=target_profile_id,
        strategy=strategy,
    )
    with self.unit_of_work:
        self.message_bus.handle(command)

message_bus instance-attribute

message_bus = message_bus or MessageBus(
    unit_of_work,
    injected_event_handlers,
    injected_command_handlers,
)

search_by_age_greater_or_equal_to

search_by_age_greater_or_equal_to(age)

Search profiles by their age and using their birthday

Source code in components/global_profile/public/api.py

@obs.api_call()
def search_by_age_greater_or_equal_to(self, age: int) -> list[Profile]:
    """Search profiles by their age and using their birthday"""
    with self.unit_of_work:
        profiles = self.unit_of_work.profiles.search_by_age_greater_or_equal_to(age)
        return cast("list[Profile]", profiles)

search_profiles_by_fullname

search_profiles_by_fullname(fullname)

Search profiles by their full name

Source code in components/global_profile/public/api.py

@obs.api_call()
def search_profiles_by_fullname(self, fullname: str) -> list[Profile]:
    """Search profiles by their full name"""
    with self.unit_of_work:
        profiles = self.unit_of_work.profiles.search_by_fullname(fullname)
        return cast("list[Profile]", profiles)

search_profiles_by_phone_number

search_profiles_by_phone_number(phone_number)

Search profiles by phone number

Source code in components/global_profile/public/api.py

@obs.api_call()
def search_profiles_by_phone_number(self, phone_number: str) -> list[Profile]:
    """Search profiles by phone number"""
    with self.unit_of_work:
        profiles = self.unit_of_work.profiles.search_by_phone_number(phone_number)
        return cast("list[Profile]", profiles)

unit_of_work instance-attribute

unit_of_work = unit_of_work or TransactionUnitOfWork()

user_compat instance-attribute

user_compat = GlobalUserCompatibility(unit_of_work)

components.global_profile.public.constants

GLOBAL_PROFILE_SCHEMA_NAME module-attribute

GLOBAL_PROFILE_SCHEMA_NAME = 'global_profile'

components.global_profile.public.entities

Address dataclass

Address(
    street,
    postal_code,
    locality,
    country,
    unit=None,
    administrative_area=None,
    sublocality=None,
    postal_town=None,
)

Represents a physical address with international support.

Attributes:

Name	Type	Description
street	str	The street name and number (e.g., "123 Main Street")
unit	str | None	Additional address details like apartment number, suite, or floor (e.g., "Apt 4B", "Suite 200")
locality	str | None	The city or town name (e.g., "Paris", "London")
administrative_area	str | None	The state, province, region, or department (e.g., "Île-de-France", "California")
postal_code	str | None	The ZIP code or postal code (e.g., "75001", "90210")
country	str	The ISO 3166-1 alpha-2 for the country.
sublocality	str | None	A smaller division within a locality such as district, borough, or neighborhood (e.g., "Le Marais", "Brooklyn")
postal_town	str | None	The post town or postal locality, which may differ from the locality (particularly relevant in some countries like the UK)

Example

address = Address(
    street="123 Main Street",
    unit="Apt 4B",
    locality="Paris",
    administrative_area="Île-de-France",
    postal_code="75001",
    country="FR",
    sublocality="Le Marais",
    postal_town=None
)

Errors

ValueError: for invalid country code or invalid administrative area code AddressInvalidException: for country specific validation e.g. Canada location and postal codes

__post_init__

__post_init__()

Enforce ISO 3166 code validation

Source code in components/global_profile/internal/domain/entities.py

def __post_init__(self) -> None:
    """
    Enforce ISO 3166 code validation
    """

    # Sanitize inputs
    sanitized_values = {
        "street": self.street.strip(),
        "country": self.country.strip(),
        "unit": self.unit.strip() if self.unit is not None else None,
        "locality": self.locality.strip() if self.locality is not None else None,
        "administrative_area": (
            self.administrative_area.strip()
            if self.administrative_area is not None
            else None
        ),
        "postal_code": (
            self.postal_code.strip() if self.postal_code is not None else None
        ),
        "sublocality": (
            self.sublocality.strip() if self.sublocality is not None else None
        ),
        "postal_town": (
            self.postal_town.strip() if self.postal_town is not None else None
        ),
    }

    # Country specific sanitize inputs
    if sanitized_values["country"] == "CA" and self.postal_code is not None:
        sanitized_values["postal_code"] = self.postal_code.replace(" ", "").upper()

    # Update instance with sanitized values
    if any(getattr(self, key) != value for key, value in sanitized_values.items()):
        for key, value in sanitized_values.items():
            object.__setattr__(self, key, value)

    # Validate country code
    if pycountry.countries.get(alpha_2=self.country) is None:
        raise ValueError(
            f"Invalid country code: {self.country}; it must be ISO 3166-1 alpha-2!"
        )
    # Validate administrative area if present
    if self.administrative_area:
        if pycountry.subdivisions.get(code=self.administrative_area) is None:  # type: ignore[no-untyped-call]
            raise ValueError(
                f"Invalid administrative area code: {self.administrative_area}; it must be ISO 3166-2!"
            )
    # Validate existence of Canadian addresses
    if self.country == "CA":
        if self._postal_code_ca_re.match(mandatory(self.postal_code)) is None:
            raise AddressInvalidException()
    elif self.country == "FR":
        if not self.locality:
            raise ValueError("Locality cannot be empty for 🇫🇷 addresses")

        if not self._postal_code_fr_re.match(mandatory(self.postal_code)):
            raise ValueError(
                f"{self.postal_code} does not match the expected postal code for 🇫🇷"
            )

administrative_area class-attribute instance-attribute

administrative_area = None

country instance-attribute

country

locality instance-attribute

locality

postal_code instance-attribute

postal_code

postal_town class-attribute instance-attribute

postal_town = None

street instance-attribute

street

sublocality class-attribute instance-attribute

sublocality = None

unit class-attribute instance-attribute

unit = None

Gender

Bases: StrEnum

Gender of a person.

Note: this is different from the biological sex.

FEMALE class-attribute instance-attribute

FEMALE = 'female'

MALE class-attribute instance-attribute

MALE = 'male'

NON_BINARY class-attribute instance-attribute

NON_BINARY = 'non_binary'

OTHER class-attribute instance-attribute

OTHER = 'other'

PREFER_NOT_TO_SAY class-attribute instance-attribute

PREFER_NOT_TO_SAY = 'prefer_not_to_say'

Language

Bases: StrEnum

Language spoken by a person.

Used to define the preferred language for communication. It is naturally constrained by the languages supported at Alan.

Must use ISO-639 nomenclature, see https://en.wikipedia.org/wiki/List_of_ISO_639-1_codes ⧉

DUTCH class-attribute instance-attribute

DUTCH = 'nl'

ENGLISH class-attribute instance-attribute

ENGLISH = 'en'

FRENCH class-attribute instance-attribute

FRENCH = 'fr'

SPANISH class-attribute instance-attribute

SPANISH = 'es'

PlaceIdentifierType

Bases: StrEnum

Available types to identify a place.

Other types that could be used (not exhaustive): NUTS = "nuts" # EU Nomenclature of Territorial Units for Statistics - hierarchical regional codes UN_LOCODE = "unlocode" # UN Location Codes - international standard for cities and ports (5 chars) WIKIDATA = "wikidata" # Wikidata knowledge base - universal place identifiers (Q-numbers) GEONAMES = "geonames" # GeoNames geographical database - numeric place identifiers

INSEE class-attribute instance-attribute

INSEE = 'insee'

PlaceOfBirth dataclass

PlaceOfBirth(name, country=None, identifiers=None)

Represents a geographical place with optional identifiers.

This class can represent any type of place from cities to countries to regions, with varying levels of precision and identification.

Attributes:

Name	Type	Description
name	str	Human-readable place name (e.g., "Paris", "Normandy", "Unknown")
country	str | None	ISO 3166-1 alpha-2 country code (e.g., "FR", "DE") or None if unknown
identifiers	dict[PlaceIdentifierType, str] | None	Dictionary mapping identifier providers to their codes, or None if no identifiers available

Examples:

>>> PlaceOfBirth("Paris", "FR").with_place_identifier(PlaceIdentifierType.INSEE, "75056")
PlaceOfBirth(name='Paris', country='FR', identifiers={<PlaceIdentifierType.INSEE: 'insee'>: '75056'})

>>> PlaceOfBirth("Somewhere in France", country="FR")
PlaceOfBirth(name='Somewhere in France', country='FR', identifiers=None)

>>> PlaceOfBirth("Unknown")
PlaceOfBirth(name='Unknown', country=None, identifiers=None)

__post_init__

__post_init__()

Source code in components/global_profile/internal/domain/entities.py

def __post_init__(self) -> None:
    # Validate country code
    if (
        self.country is not None
        and pycountry.countries.get(alpha_2=self.country) is None
    ):
        raise ValueError(
            f"Invalid country code: {self.country}; it must be ISO 3166-1 alpha-2!"
        )

    # Validate identifiers
    if self.identifiers is not None:
        for identifier_type, code in self.identifiers.items():
            match identifier_type:
                case PlaceIdentifierType.INSEE:
                    if not (len(code) == 5 and code.isdigit()):
                        raise ValueError(
                            f"Invalid INSEE identifier: {code}; it must be a 5-digit number!"
                        )

country class-attribute instance-attribute

country = None

get_place_identifier

get_place_identifier(identifier_type)

Returns the identifier code for the given identifier type, or None if not found.

Source code in components/global_profile/internal/domain/entities.py

def get_place_identifier(self, identifier_type: PlaceIdentifierType) -> str | None:
    """Returns the identifier code for the given identifier type, or None if not found."""
    return self.identifiers.get(identifier_type) if self.identifiers else None

identifiers class-attribute instance-attribute

identifiers = None

name instance-attribute

name

with_place_identifier

with_place_identifier(identifier_type, code)

Returns a new instance, adding or replacing the identifier_type code.

Source code in components/global_profile/internal/domain/entities.py

def with_place_identifier(
    self, identifier_type: PlaceIdentifierType, code: str
) -> Self:
    """Returns a new instance, adding or replacing the identifier_type code."""
    new_identifiers = dict(self.identifiers) if self.identifiers else {}
    new_identifiers[identifier_type] = code
    return self.__class__(self.name, self.country, new_identifiers)

Profile

Bases: Protocol

Profile public type interface of the ProfileService.

additional_first_names instance-attribute

additional_first_names

address instance-attribute

address

age property

age

Returns the age of a Profile on the current date.

age_on

age_on(age_date)

Returns the age of a Profile on a given date.

Source code in components/global_profile/public/entities.py

def age_on(self, age_date: date) -> float | None:
    """
    Returns the age of a Profile on a given date.
    """
    ...

birth_date instance-attribute

birth_date

birth_name instance-attribute

birth_name

consents instance-attribute

consents

email instance-attribute

email

first_name instance-attribute

first_name

full_name property

full_name

Returns all names parts of a Profile in a single string.

gender instance-attribute

gender

id instance-attribute

id

last_name instance-attribute

last_name

phone_number instance-attribute

phone_number

place_of_birth instance-attribute

place_of_birth

preferred_language instance-attribute

preferred_language

ProfileConsents dataclass

ProfileConsents(values=list())

Bases: DataClassJsonMixin

Collected consents

values class-attribute instance-attribute

values = field(default_factory=list)

ProfileMergeStrategy

Bases: StrEnum

Strategies to merge two profiles. - SOURCE_DELETED: the source profile is deleted and the target profile remains unchanged. - TARGET_WINS: the target profile's non-null fields take precedence over the source profile's fields. (target_profile.field = target_profile.field or source_profile.field). The source profile is then deleted.

SOURCE_DELETED class-attribute instance-attribute

SOURCE_DELETED = 'source_deleted'

TARGET_WINS class-attribute instance-attribute

TARGET_WINS = 'target_wins'

components.global_profile.public.events

BirthInformationChanged dataclass

BirthInformationChanged(
    profile_id, birth_date, birth_name, place_of_birth
)

Bases: Message

BirthInformationChanged event

birth_date instance-attribute

birth_date

birth_name instance-attribute

birth_name

place_of_birth instance-attribute

place_of_birth

profile_id instance-attribute

profile_id

ConsentsChanged dataclass

ConsentsChanged(consents, profile_id)

Bases: Message

ConsentsChanged event

consents instance-attribute

consents

profile_id instance-attribute

profile_id

IdentityInformationChanged dataclass

IdentityInformationChanged(
    profile_id,
    first_name,
    last_name,
    additional_first_names=None,
    birth_date=None,
)

Bases: Message

IdentityInformationChanged event

additional_first_names class-attribute instance-attribute

additional_first_names = None

birth_date class-attribute instance-attribute

birth_date = None

first_name instance-attribute

first_name

last_name instance-attribute

last_name

profile_id instance-attribute

profile_id

PreferredLanguageChanged dataclass

PreferredLanguageChanged(profile_id, preferred_language)

Bases: Message

ProfilePreferredLanguageChanged event

preferred_language instance-attribute

preferred_language

profile_id instance-attribute

profile_id

ProfileAddressChanged dataclass

ProfileAddressChanged(profile_id, address)

Bases: Message

AddressChanged event

address instance-attribute

address

profile_id instance-attribute

profile_id

ProfileEmailChanged dataclass

ProfileEmailChanged(profile_id, email)

Bases: Message

EmailChanged event

email instance-attribute

email

profile_id instance-attribute

profile_id

ProfilesMerged dataclass

ProfilesMerged(source_profile_id, target_profile_id)

Bases: Message

ProfilesMerged event

source_profile_id instance-attribute

source_profile_id

target_profile_id instance-attribute

target_profile_id

components.global_profile.public.exceptions

AddressInvalidException

AddressInvalidException()

Bases: Exception

Source code in components/global_profile/internal/domain/exceptions.py

def __init__(self) -> None:
    super().__init__("Address invalid")

ProfileWithEmailAlreadyExistsException

ProfileWithEmailAlreadyExistsException()

Bases: Exception

Source code in components/global_profile/internal/domain/exceptions.py

def __init__(self) -> None:
    super().__init__("Profile with email already exists")

UserMissingProfileException

UserMissingProfileException(user_id)

Bases: Exception

Source code in components/global_profile/internal/domain/exceptions.py

def __init__(self, user_id: UUID) -> None:
    super().__init__(f"User '{user_id}' is missing a profile")

UserNotFoundForProfileException

UserNotFoundForProfileException(profile_id)

Bases: Exception

Source code in components/global_profile/internal/domain/exceptions.py

def __init__(self, profile_id: UUID) -> None:
    super().__init__(f"No user found for profile ID {profile_id}")

components.global_profile.public.inject_profile_service

T module-attribute

T = TypeVar('T')

inject_profile_service

inject_profile_service(f)

Inject a ProfileService in method parameters.

This injector mimics a DI injection of the ProfileService to keep the door opened for a DI framework if needed

Source code in components/global_profile/public/inject_profile_service.py

def inject_profile_service(f: Callable[..., T]) -> Callable[..., T]:
    """
    Inject a ProfileService in method parameters.

    This injector mimics a DI injection of the ProfileService to keep the door opened for a DI framework if needed
    """
    from components.global_profile.public.api import ProfileService

    method_signature = signature(f)

    @wraps(f)
    def decorated_function(*args, **kwargs):  # type: ignore[no-untyped-def]
        kwargs["profile_service"] = ProfileService.create()
        return f(*args, **kwargs)

    if "profile_service" in method_signature.parameters:
        return decorated_function
    return f

components.global_profile.public.mappers

ENTITYT module-attribute

ENTITYT = TypeVar('ENTITYT')

GenderMapper

Bases: Mapper[UserGender | None, Gender | None]

Maps between UserGender|None<>Gender|None

to_entity staticmethod

to_entity(gender)

Source code in components/global_profile/public/mappers.py

@staticmethod
@override
def to_entity(gender: UserGender | None) -> Gender | None:
    match gender:
        case None:
            return None
        case UserGender.male:
            return Gender.MALE
        case UserGender.female:
            return Gender.FEMALE
        case UserGender.non_binary:
            return Gender.NON_BINARY
        case _:
            raise NotImplementedError(f"Unknown gender {gender}")

to_model staticmethod

to_model(gender)

Source code in components/global_profile/public/mappers.py

@staticmethod
@override
def to_model(gender: Gender | None) -> UserGender | None:
    match gender:
        case None | Gender.NON_BINARY:
            return None
        case Gender.MALE:
            return UserGender.male
        case Gender.FEMALE:
            return UserGender.female
        case Gender.OTHER:
            return None
        case Gender.PREFER_NOT_TO_SAY:
            return None
        case _:
            raise NotImplementedError(f"Unknown gender {gender}")

LanguageMapper

Bases: Mapper[Lang, Language]

Maps between Lang<>Language

to_entity staticmethod

to_entity(lang)

Source code in components/global_profile/public/mappers.py

@staticmethod
@override
def to_entity(lang: Lang) -> Language:
    return Language(lang)

to_model staticmethod

to_model(language)

Source code in components/global_profile/public/mappers.py

@staticmethod
@override
def to_model(language: Language) -> Lang:
    return Lang(language)

MODELT module-attribute

MODELT = TypeVar('MODELT')

Mapper

Bases: ABC, Generic[MODELT, ENTITYT]

Maps between MODELT<>ENTITYT

to_entity abstractmethod staticmethod

to_entity(model)

Maps from MODELT to ENTITTYT

Source code in components/global_profile/public/mappers.py

@staticmethod
@abstractmethod
def to_entity(model: MODELT) -> ENTITYT:
    """Maps from MODELT to ENTITTYT"""

to_model abstractmethod staticmethod

to_model(entity)

Maps from ENTITYT to MODELT

Source code in components/global_profile/public/mappers.py

@staticmethod
@abstractmethod
def to_model(entity: ENTITYT) -> MODELT:
    """Maps from ENTITYT to MODELT"""

ProfileMapper

Bases: Mapper[MODELT, Profile]

Maps between MODELT<>Profile

components.global_profile.public.profile_column_decorator

T module-attribute

T = TypeVar('T', bound=UUID | (UUID | None))

global_profile_link

global_profile_link(func)

Use this decorator to mark a UUID column as a global profile id following the example bellow. If the id of the original were to be merged with another, the marked column would be migrated. ⚠️ always the decorator declared_attr on top of this decorator (and not the other way around). ⚠️ it does not work for columns with constraints.

@declared_attr
@global_profile_link
def profile_id(cls) -> Mapped[uuid.UUID]:
    return mapped_column(PostgreSQLUUID(as_uuid=True))

Source code in components/global_profile/public/profile_column_decorator.py

def global_profile_link(
    func: Callable[..., Mapped[T]],
) -> Callable[..., Mapped[T]]:
    """
    Use this decorator to mark a UUID column as a global profile id following the example bellow.
    If the id of the original were to be merged with another, the marked column would be migrated.
    ⚠️ always the decorator `declared_attr` on top of this decorator (and not the other way around).
    ⚠️ it does not work for columns with constraints.
    ```
    @declared_attr
    @global_profile_link
    def profile_id(cls) -> Mapped[uuid.UUID]:
        return mapped_column(PostgreSQLUUID(as_uuid=True))
    ```
    """

    @wraps(func)
    def wrapper(cls: BaseModel) -> Mapped[T]:
        # Call the original function to get the column
        column = func(cls)
        if isinstance(column, MappedColumn):
            column = column.column  # type: ignore[assignment]

        # Add metadata to the column's info dictionary
        if not hasattr(column, "info") or column.info is None:
            column.info = {}  # type: ignore[attr-defined]

        column.info["is_global_profile_link"] = True  # type: ignore[attr-defined]

        return column

    return wrapper

components.global_profile.public.subscriptions

subscribe_to_events

subscribe_to_events()

All event subscriptions for the global profile component should be done here.

Source code in components/global_profile/public/subscriptions.py

def subscribe_to_events() -> None:
    """
    All event subscriptions for the global profile component should be done here.
    """
    from components.global_profile.internal.events.subscribers import (
        update_links_to_profile_when_merging,
    )
    from components.global_profile.public.events import ProfilesMerged
    from shared.messaging.broker import get_message_broker
    from shared.queuing.config import LOW_PRIORITY_QUEUE

    message_broker = get_message_broker()

    #  Once bootstraped by a given app, this subscription will make sure that, when a profile is merged into
    #  another, the profile ids are updated accordingly
    message_broker.subscribe_async(
        ProfilesMerged,
        update_links_to_profile_when_merging,
        queue_name=LOW_PRIORITY_QUEUE,
    )

components.global_profile.public.transformers

GenderTransformer

Transforms different gender representations to and from the domain model.

The domain model (Gender enum) represents gender biaised towards gender/biological sex distinction, while external representations might have different meanings.

to_domain staticmethod

to_domain(gender: Gender | UserGender | str) -> Gender

to_domain(gender: None) -> None

to_domain(gender)

Transforms UserGender or str representation to domain model.

If str input is used, must valid string representations of Gender or UserGender

Source code in components/global_profile/public/transformers.py

@staticmethod
def to_domain(gender: Gender | UserGender | str | None) -> Gender | None:
    """
    Transforms UserGender or str representation to domain model.

    If str input is used, must valid string representations of Gender or UserGender
    """
    if gender is None:
        return None

    match gender:
        case Gender(value):
            return Gender(value)
        case UserGender.male:
            return Gender.MALE
        case UserGender.female:
            return Gender.FEMALE
        case UserGender.non_binary:
            return Gender.NON_BINARY
        case "M":
            return Gender.MALE
        case "F":
            return Gender.FEMALE
        case "X":
            return Gender.NON_BINARY
        case _:
            return Gender(gender)

LanguageTransformer

Transforms different language representations to domain Language.

to_domain staticmethod

to_domain(language)

Transforms Lang or str representation to domain model.

If str input is used, must be a ISO 639-2 code

Source code in components/global_profile/public/transformers.py

@staticmethod
def to_domain(language: Language | Lang | str) -> Language:
    """
    Transforms Lang or str representation to domain model.

    If str input is used, must be a ISO 639-2 code
    """
    match language:
        case Language(value):
            return Language(value)
        case Lang(value):
            return Language(value)
        case _:
            return Language(language)