mex.extractors.consent_mailer package

Submodules

mex.extractors.consent_mailer.extract module

mex.extractors.consent_mailer.extract.extract_consents_for_persons(person_items: list[MergedPerson]) → list[MergedConsent]

Get consents for ldap persons.

mex.extractors.consent_mailer.extract.extract_ldap_persons(extracted_primary_source_ldap_identifier: MergedPrimarySourceIdentifier) → list[MergedPerson]

Get all persons from primary source LDAP.

mex.extractors.consent_mailer.filter module

mex.extractors.consent_mailer.filter.filter_persons_without_consent(person_items: list[MergedPerson], consent_items: list[MergedConsent]) → list[MergedPerson]

Filter person items for having no consent.

Parameters:

• person_items – list of persons

• consent_items – list of consents

Returns:

list of filtered persons without consent.

mex.extractors.consent_mailer.main module

mex.extractors.consent_mailer.settings module

class mex.extractors.consent_mailer.settings.ConsentMailerSettings(_env_file: Path | str | Sequence[Path | str] | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_nested_delimiter: str | None = None, _secrets_dir: str | Path | None = None, *, pdb: bool = False, MEX_SINK: list[Sink] = [Sink.NDJSON], MEX_ASSETS_DIR: Path = PosixPath('/home/runner/work/mex-extractors/mex-extractors/assets'), MEX_WORK_DIR: Path = PosixPath('/home/runner/work/mex-extractors/mex-extractors'), MEX_IDENTITY_PROVIDER: IdentityProvider = IdentityProvider.MEMORY, MEX_BACKEND_API_URL: Url = Url('http://localhost:8080/'), MEX_BACKEND_API_KEY: SecretStr = SecretStr('**********'), MEX_BACKEND_API_PARALLELIZATION: int = 1, MEX_BACKEND_API_CHUNK_SIZE: int = 25, MEX_VERIFY_SESSION: bool | AssetsPath = True, MEX_ORGANIGRAM_PATH: AssetsPath = AssetsPath('raw-data/organigram/organizational_units.json'), MEX_PRIMARY_SOURCES_PATH: AssetsPath = AssetsPath('raw-data/primary-sources/primary-sources.json'), MEX_LDAP_URL: SecretStr = SecretStr('**********'), MEX_LDAP_SEARCH_BASE: str = 'DC=rki,DC=local', MEX_WIKI_API_URL: Url = Url('http://wikidata/'), MEX_WEB_USER_AGENT: str = 'rki/mex', MEX_ORCID_API_URL: Url = Url('https://orcid/'), mailpit_api_url: str = 'localhost:8025', mailpit_api_user: SecretStr = SecretStr('**********'), mailpit_api_password: SecretStr = SecretStr('**********'), schedule: str | None = None, smtp_server: str = 'localhost:1025', template_path: AssetsPath = AssetsPath('mailings'))

Bases: BaseSettings

Settings definition class for the consent mailer.

mailpit_api_password: SecretStr

mailpit_api_url: str

mailpit_api_user: SecretStr

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': '.env', 'env_file_encoding': 'utf-8', 'env_ignore_empty': False, 'env_nested_delimiter': '__', 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'mex_', 'extra': 'ignore', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'populate_by_name': True, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_assignment': True, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'assets_dir': FieldInfo(annotation=Path, required=False, default=PosixPath('/home/runner/work/mex-extractors/mex-extractors/assets'), alias_priority=2, validation_alias='MEX_ASSETS_DIR', description='Path to directory that contains input files treated as read-only, looks for a folder named `assets` in the current directory by default.'), 'backend_api_chunk_size': FieldInfo(annotation=int, required=False, default=25, alias_priority=2, validation_alias='MEX_BACKEND_API_CHUNK_SIZE', description='How many items to load into the backend in one chunk.'), 'backend_api_key': FieldInfo(annotation=SecretStr, required=False, default=SecretStr('**********'), alias_priority=2, validation_alias='MEX_BACKEND_API_KEY', description='Backend API key with write access to call POST/PUT endpoints'), 'backend_api_parallelization': FieldInfo(annotation=int, required=False, default=1, alias_priority=2, validation_alias='MEX_BACKEND_API_PARALLELIZATION', description='How many simultaneous threads may spin up to load data into the backend.'), 'backend_api_url': FieldInfo(annotation=Url, required=False, default=Url('http://localhost:8080/'), alias_priority=2, validation_alias='MEX_BACKEND_API_URL', description='MEx backend API url.'), 'debug': FieldInfo(annotation=bool, required=False, default=False, alias='pdb', alias_priority=2, validation_alias='MEX_DEBUG', description='Jump into post-mortem debugging after any uncaught exception.'), 'identity_provider': FieldInfo(annotation=IdentityProvider, required=False, default=<IdentityProvider.MEMORY: 'memory'>, alias_priority=2, validation_alias='MEX_IDENTITY_PROVIDER', description='Provider to assign identifiers to new model instances.'), 'ldap_search_base': FieldInfo(annotation=str, required=False, default='DC=rki,DC=local', alias_priority=2, validation_alias='MEX_LDAP_SEARCH_BASE', description='Search base for the ldap connector.'), 'ldap_url': FieldInfo(annotation=SecretStr, required=False, default=SecretStr('**********'), alias_priority=2, validation_alias='MEX_LDAP_URL', description='LDAP server for person queries with authentication credentials. Must follow format `ldap://user:pw@host:port`, where `user` is the username, and `pw` is the password for authenticating against ldap, `host` is the url of the ldap server, and `port` is the port of the ldap server.'), 'mailpit_api_password': FieldInfo(annotation=SecretStr, required=False, default=SecretStr('**********'), description='The password used for Basic HTTP auth against the mailpit api. USED FOR TESTS ONLY!'), 'mailpit_api_url': FieldInfo(annotation=str, required=False, default='localhost:8025', description='The url to the api endpoint for mailpit. USED FOR TESTS ONLY!'), 'mailpit_api_user': FieldInfo(annotation=SecretStr, required=False, default=SecretStr('**********'), description='The username used for Basic HTTP auth against the mailpit api. USED FOR TESTS ONLY!'), 'mex_web_user_agent': FieldInfo(annotation=str, required=False, default='rki/mex', alias_priority=2, validation_alias='MEX_WEB_USER_AGENT', description='User agent is sent in request headers to external services.'), 'orcid_api_url': FieldInfo(annotation=Url, required=False, default=Url('https://orcid/'), alias_priority=2, validation_alias='MEX_ORCID_API_URL', description='URL of orcid api.'), 'organigram_path': FieldInfo(annotation=AssetsPath, required=False, default=AssetsPath("raw-data/organigram/organizational_units.json"), alias_priority=2, validation_alias='MEX_ORGANIGRAM_PATH', description='Path to the JSON file describing the organizational units, absolute path or relative to `assets_dir`.'), 'primary_sources_path': FieldInfo(annotation=AssetsPath, required=False, default=AssetsPath("raw-data/primary-sources/primary-sources.json"), alias_priority=2, validation_alias='MEX_PRIMARY_SOURCES_PATH', description='Path to the JSON file describing the primary sources, absolute path or relative to `assets_dir`.'), 'schedule': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='A valid cron string defining when to run the consent mailer'), 'sink': FieldInfo(annotation=list[Sink], required=False, default=[<Sink.NDJSON: 'ndjson'>], alias_priority=2, validation_alias='MEX_SINK', description='Where to send data that is extracted or ingested. Defaults to writing ndjson files, but can be configured to push to the backend or the graph.'), 'smtp_server': FieldInfo(annotation=str, required=False, default='localhost:1025', description='Address and port (<address>:<port>) of the SMTP server to use for the consent mailer.'), 'template_path': FieldInfo(annotation=AssetsPath, required=False, default=AssetsPath("mailings"), description='Path to the directory with the jinja template file containing the email body template, absolute path or relative to `assets_dir`.'), 'verify_session': FieldInfo(annotation=Union[bool, AssetsPath], required=False, default=True, alias_priority=2, validation_alias='MEX_VERIFY_SESSION', description="Either a boolean that controls whether we verify the server's TLS certificate, or a path to a CA bundle to use. If a path is given, it can be either absolute or relative to the `assets_dir`. Defaults to True."), 'wiki_api_url': FieldInfo(annotation=Url, required=False, default=Url('http://wikidata/'), alias_priority=2, validation_alias='MEX_WIKI_API_URL', description='URL of the Wikidata API used to resolve an ID to an organization.'), 'work_dir': FieldInfo(annotation=Path, required=False, default=PosixPath('/home/runner/work/mex-extractors/mex-extractors'), alias_priority=2, validation_alias='MEX_WORK_DIR', description='Path to directory that stores generated and temporary files. Defaults to the current working directory.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

schedule: str | None

smtp_server: str

template_path: AssetsPath

mex.extractors.consent_mailer.transform module

mex.extractors.consent_mailer.transform._generate_email_body(person: MergedPerson, template: Template, template_args: dict[str, Any]) → str

Generates the consent email body.

Parameters:

• person – person to send email to

• template – email template

• template_args – dict with args inserted into template

Returns:

email body string

mex.extractors.consent_mailer.transform.transform_person_to_sendable_email(person: MergedPerson) → EmailMessage | None

Transforms a person to an email requesting consent.

Following properties are used: - the subject defined in ‘settings.consent_mailer.template_path / “config.json”’ - the person email addresses (ending on @rki.de) as target addresses (if none is present, this function will return none) - the template defined in ‘settings.consent_mailer.template_path / “consent.html”’ to generate the email body (text) for each person

Parameters:

• person (MergedPerson) – The person that needs to be transformed into a sendable

• email. (consent)

Returns:

The email message for the defined person or None if the person doesn’t have a @rki.de email address.

Return type:

EmailMessage | None

Module contents