Queries

components.payment_gateway.subcomponents.parties.protected.business_logic.queries.financial_instrument_provider_mapping_queries

FinancialInstrumentProviderMappingQueries

Queries for financial instrument provider mapping operations.

get_mapping_by_financial_instrument_and_workspace classmethod

get_mapping_by_financial_instrument_and_workspace(
    session, /, financial_instrument_id, workspace_key
)

Get provider mapping by financial instrument ID and workspace key.

Parameters:

Name	Type	Description	Default
session	Session	Database session	required
financial_instrument_id	FinancialInstrumentId	Financial instrument ID	required
workspace_key	str	Provider workspace key	required

Returns:

Type	Description
FinancialInstrumentProviderMapping | None	The provider mapping or None

Source code in components/payment_gateway/subcomponents/parties/protected/business_logic/queries/financial_instrument_provider_mapping_queries.py

@classmethod
def get_mapping_by_financial_instrument_and_workspace(
    cls,
    session: Session,
    /,
    financial_instrument_id: FinancialInstrumentId,
    workspace_key: str,
) -> FinancialInstrumentProviderMapping | None:
    """Get provider mapping by financial instrument ID and workspace key.

    Args:
        session: Database session
        financial_instrument_id: Financial instrument ID
        workspace_key: Provider workspace key

    Returns:
        The provider mapping or None
    """
    return FinancialInstrumentProviderMappingModelBroker.get_mapping_by_financial_instrument_and_workspace(
        session,
        financial_instrument_id=financial_instrument_id,
        workspace_key=workspace_key,
    )

components.payment_gateway.subcomponents.parties.protected.business_logic.queries.financial_instrument_queries

FinancialInstrumentQueries

Queries for financial instrument operations.

get_financial_instrument classmethod

get_financial_instrument(
    session,
    /,
    financial_instrument_id,
    with_legal_entity=False,
)

Get a financial instrument by ID.

Parameters:

Name	Type	Description	Default
session	Session	Database session	required
financial_instrument_id	FinancialInstrumentId	Financial instrument ID	required
with_legal_entity	bool	Whether to eager load the legal entity relationship	False

Returns:

Type	Description
FinancialInstrument	The financial instrument

Raises:

Type	Description
FinancialInstrumentNotFoundException	If the financial instrument is not found

Source code in components/payment_gateway/subcomponents/parties/protected/business_logic/queries/financial_instrument_queries.py

@classmethod
def get_financial_instrument(
    cls,
    session: Session,
    /,
    financial_instrument_id: FinancialInstrumentId,
    with_legal_entity: bool = False,
) -> FinancialInstrument:
    """Get a financial instrument by ID.

    Args:
        session: Database session
        financial_instrument_id: Financial instrument ID
        with_legal_entity: Whether to eager load the legal entity relationship

    Returns:
        The financial instrument

    Raises:
        FinancialInstrumentNotFoundException: If the financial instrument is not found
    """
    with raise_if_financial_instrument_not_found(financial_instrument_id):
        return FinancialInstrumentModelBroker.get_financial_instrument(
            session,
            id=financial_instrument_id,
            with_legal_entity=with_legal_entity,
        )

components.payment_gateway.subcomponents.parties.protected.business_logic.queries.financial_instrument_reveal_queries

FinancialInstrumentRevealQueries

FinancialInstrumentRevealQueries(cache)

Queries for revealing encrypted financial instrument data.

Implements a two-step reveal flow with cache-based key registration: 1. Consumer registers an ephemeral RSA public key with reason/actor 2. Consumer calls reveal with the key ID to get JWE-encrypted data

Implements the Nullable pattern: - Nullables: https://www.jamesshore.com/v2/projects/nullables/testing-without-mocks#nullables ⧉ - Parameterless instantiation: https://www.jamesshore.com/v2/projects/nullables/testing-without-mocks#instantiation ⧉

Source code in components/payment_gateway/subcomponents/parties/protected/business_logic/queries/financial_instrument_reveal_queries.py

def __init__(
    self,
    cache: "AlanCache",
) -> None:
    self._cache = cache

create classmethod

create()

Normal factory.

Source code in components/payment_gateway/subcomponents/parties/protected/business_logic/queries/financial_instrument_reveal_queries.py

@classmethod
def create(cls) -> "FinancialInstrumentRevealQueries":
    """Normal factory."""
    return cls(cache=alan_cache)

create_null classmethod

create_null()

Null factory for testing.

Source code in components/payment_gateway/subcomponents/parties/protected/business_logic/queries/financial_instrument_reveal_queries.py

@classmethod
def create_null(cls) -> "FinancialInstrumentRevealQueries":
    """Null factory for testing."""
    return cls(cache=alan_cache)

register_reveal_key

register_reveal_key(public_key_pem, *, reason, actor)

Register an ephemeral RSA public key for a reveal operation.

The key is stored in cache with a short TTL and will be consumed (deleted) on the first reveal call that uses it.

Parameters:

Name	Type	Description	Default
public_key_pem	str	RSA public key in PEM format.	required
reason	str	Why the reveal is needed (free-form).	required
actor	str	Who is requesting the reveal (e.g. "component:banking_documents").	required

Returns:

Type	Description
str	A unique reveal_key_id to use in the subsequent reveal call.

Source code in components/payment_gateway/subcomponents/parties/protected/business_logic/queries/financial_instrument_reveal_queries.py

def register_reveal_key(
    self,
    public_key_pem: str,
    *,
    reason: str,
    actor: str,
) -> str:
    """Register an ephemeral RSA public key for a reveal operation.

    The key is stored in cache with a short TTL and will be consumed
    (deleted) on the first reveal call that uses it.

    Args:
        public_key_pem: RSA public key in PEM format.
        reason: Why the reveal is needed (free-form).
        actor: Who is requesting the reveal (e.g. "component:banking_documents").

    Returns:
        A unique reveal_key_id to use in the subsequent reveal call.
    """
    reveal_key_id = str(uuid.uuid4())
    cache_key = f"{_REVEAL_KEY_CACHE_PREFIX}{reveal_key_id}"

    cache_value = json.dumps(
        {
            "public_key_pem": public_key_pem,
            "reason": reason,
            "actor": actor,
        }
    )

    self._cache.set(cache_key, cache_value, _REVEAL_KEY_TTL)

    current_logger.info(
        "Reveal key registered",
        reveal_key_id=reveal_key_id,
        reason=reason,
        actor=actor,
    )

    return reveal_key_id

reveal_ca_local_account_details

reveal_ca_local_account_details(
    session, /, id, reveal_key_id
)

Reveal Canadian local account details as a JWE token.

Fetches and deletes the registered public key from cache (single-use), decrypts the CA local account data from the database, and re-encrypts it as a JWE token using the consumer's public key.

Parameters:

Name	Type	Description	Default
session	Session	Database session.	required
id	FinancialInstrumentId	Financial instrument ID.	required
reveal_key_id	str	Key ID returned by register_reveal_key.	required

Returns:

Type	Description
str	JWE token string containing the encrypted CALocalAccountDetails.

Raises:

Type	Description
RevealKeyNotFoundException	If the key ID is not found (expired or used).
FinancialInstrumentNotFoundException	If the FI doesn't exist.
FinancialInstrumentTypeNotSupportedException	If the FI is not a CA local account.

Source code in components/payment_gateway/subcomponents/parties/protected/business_logic/queries/financial_instrument_reveal_queries.py

def reveal_ca_local_account_details(
    self,
    session: Session,
    /,
    id: FinancialInstrumentId,
    reveal_key_id: str,
) -> str:
    """Reveal Canadian local account details as a JWE token.

    Fetches and deletes the registered public key from cache (single-use),
    decrypts the CA local account data from the database, and re-encrypts it
    as a JWE token using the consumer's public key.

    Args:
        session: Database session.
        id: Financial instrument ID.
        reveal_key_id: Key ID returned by register_reveal_key.

    Returns:
        JWE token string containing the encrypted CALocalAccountDetails.

    Raises:
        RevealKeyNotFoundException: If the key ID is not found (expired or used).
        FinancialInstrumentNotFoundException: If the FI doesn't exist.
        FinancialInstrumentTypeNotSupportedException: If the FI is not a CA local account.
    """
    from components.payment_gateway.subcomponents.parties.models.ca_local_account_financial_instrument import (
        CALocalAccountFinancialInstrument,
    )

    public_key_pem, reason, actor = self._consume_reveal_key(reveal_key_id)

    with raise_if_financial_instrument_not_found(id):
        fi = FinancialInstrumentModelBroker.get_financial_instrument(session, id=id)

    raise_on_financial_instrument_type_not_supported(
        fi.instrument_type,
        [FinancialInstrumentType.CA_LOCAL_ACCOUNT],
    )
    assert isinstance(fi, CALocalAccountFinancialInstrument)

    # Decrypt from DB and build the details dataclass
    ca_data = fi.ca_local_account_data_decrypted
    details = CALocalAccountDetails(
        institution_number=ca_data.institution_number,
        transit_number=ca_data.transit_number,
        account_number=ca_data.account_number,
        display_value=format_ca_local_account(
            ca_data.institution_number,
            ca_data.transit_number,
            ca_data.account_number,
        ),
    )

    # Encrypt as JWE for the consumer
    jwe_token = self._encrypt_as_jwe(details.to_json(), public_key_pem)

    current_logger.info(
        "Financial instrument data revealed",
        financial_instrument_id=str(id),
        instrument_type=FinancialInstrumentType.CA_LOCAL_ACCOUNT.value,
        reveal_key_id=reveal_key_id,
        reason=reason,
        actor=actor,
    )

    return jwe_token

reveal_iban_account_details

reveal_iban_account_details(session, /, id, reveal_key_id)

Reveal IBAN account details as a JWE token.

Fetches and deletes the registered public key from cache (single-use), decrypts the IBAN data from the database, and re-encrypts it as a JWE token using the consumer's public key.

Parameters:

Name	Type	Description	Default
session	Session	Database session.	required
id	FinancialInstrumentId	Financial instrument ID.	required
reveal_key_id	str	Key ID returned by register_reveal_key.	required

Returns:

Type	Description
str	JWE token string containing the encrypted IBANAccountDetails.

Raises:

Type	Description
RevealKeyNotFoundException	If the key ID is not found (expired or used).
FinancialInstrumentNotFoundException	If the FI doesn't exist.
FinancialInstrumentTypeNotSupportedException	If the FI is not an IBAN account.

Source code in components/payment_gateway/subcomponents/parties/protected/business_logic/queries/financial_instrument_reveal_queries.py

def reveal_iban_account_details(
    self,
    session: Session,
    /,
    id: FinancialInstrumentId,
    reveal_key_id: str,
) -> str:
    """Reveal IBAN account details as a JWE token.

    Fetches and deletes the registered public key from cache (single-use),
    decrypts the IBAN data from the database, and re-encrypts it as a JWE
    token using the consumer's public key.

    Args:
        session: Database session.
        id: Financial instrument ID.
        reveal_key_id: Key ID returned by register_reveal_key.

    Returns:
        JWE token string containing the encrypted IBANAccountDetails.

    Raises:
        RevealKeyNotFoundException: If the key ID is not found (expired or used).
        FinancialInstrumentNotFoundException: If the FI doesn't exist.
        FinancialInstrumentTypeNotSupportedException: If the FI is not an IBAN account.
    """
    from components.payment_gateway.subcomponents.parties.models.iban_account_financial_instrument import (
        IBANAccountFinancialInstrument,
    )

    public_key_pem, reason, actor = self._consume_reveal_key(reveal_key_id)

    with raise_if_financial_instrument_not_found(id):
        fi = FinancialInstrumentModelBroker.get_financial_instrument(session, id=id)

    raise_on_financial_instrument_type_not_supported(
        fi.instrument_type,
        [FinancialInstrumentType.IBAN_ACCOUNT],
    )
    assert isinstance(fi, IBANAccountFinancialInstrument)

    # Decrypt from DB and build the details dataclass
    iban_data = fi.iban_account_data_decrypted
    details = IBANAccountDetails(
        iban=iban_data.iban,
        bank_country_code=iban_data.bank_country_code,
        bic=iban_data.bic,
        display_value=format_iban(iban_data.iban),
    )

    # Encrypt as JWE for the consumer
    jwe_token = self._encrypt_as_jwe(details.to_json(), public_key_pem)

    current_logger.info(
        "Financial instrument data revealed",
        financial_instrument_id=str(id),
        instrument_type=FinancialInstrumentType.IBAN_ACCOUNT.value,
        reveal_key_id=reveal_key_id,
        reason=reason,
        actor=actor,
    )

    return jwe_token