Queries

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

FinancialInstrumentQueries

FinancialInstrumentQueries()

Queries for financial instrument operations.

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

def __init__(self) -> None:
    pass

create classmethod

create()

Normal factory.

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

@classmethod
def create(cls) -> "FinancialInstrumentQueries":
    """Normal factory."""
    return cls()

create_null classmethod

create_null()

Null factory.

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

@classmethod
def create_null(cls) -> "FinancialInstrumentQueries":
    """Null factory."""
    return cls()

get_financial_instrument

get_financial_instrument(
    session, /, financial_instrument_id
)

Get a financial instrument by ID.

Parameters:

Name	Type	Description	Default
session	Session	Database session	required
financial_instrument_id	FinancialInstrumentId	Financial instrument ID	required

Returns:

Type	Description
FinancialInstrument	The financial instrument

Raises:

Type	Description
FinancialInstrumentNotFoundException	If not found

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

def get_financial_instrument(
    self,
    session: Session,
    /,
    financial_instrument_id: FinancialInstrumentId,
) -> FinancialInstrument:
    """Get a financial instrument by ID.

    Args:
        session: Database session
        financial_instrument_id: Financial instrument ID

    Returns:
        The financial instrument

    Raises:
        FinancialInstrumentNotFoundException: If not found
    """
    with raise_if_financial_instrument_not_found(financial_instrument_id):
        model = FinancialInstrumentModelBroker.get_financial_instrument(
            session,
            id=financial_instrument_id,
        )
    return _to_financial_instrument(model)

get_financial_instrument_provider_mapping

get_financial_instrument_provider_mapping(
    session, /, financial_instrument_id, workspace_key
)

Get provider mapping for a financial instrument in a workspace.

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_queries.py

def get_financial_instrument_provider_mapping(
    self,
    session: Session,
    /,
    financial_instrument_id: FinancialInstrumentId,
    workspace_key: str,
) -> FinancialInstrumentProviderMapping | None:
    """Get provider mapping for a financial instrument in a workspace.

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

    Returns:
        The provider mapping or None
    """
    model = FinancialInstrumentProviderMappingModelBroker.get_provider_mapping_for_workspace(
        session,
        financial_instrument_id=financial_instrument_id,
        workspace_key=workspace_key,
    )
    if model is None:
        return None
    return _to_financial_instrument_provider_mapping(model)

list_financial_instruments_for_legal_entity

list_financial_instruments_for_legal_entity(
    session,
    /,
    legal_entity_id,
    *,
    instrument_type=None,
    with_terminated=False,
)

List financial instruments belonging to a legal entity.

Parameters:

Name	Type	Description	Default
session	Session	Database session	required
legal_entity_id	LegalEntityId	Legal entity ID	required
instrument_type	FinancialInstrumentType | None	Filter by instrument type (None = all types)	None
with_terminated	bool	Whether to include terminated instruments	False

Returns:

Type	Description
list[FinancialInstrument]	List of financial instruments

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

def list_financial_instruments_for_legal_entity(
    self,
    session: Session,
    /,
    legal_entity_id: LegalEntityId,
    *,
    instrument_type: FinancialInstrumentType | None = None,
    with_terminated: bool = False,
) -> list[FinancialInstrument]:
    """List financial instruments belonging to a legal entity.

    Args:
        session: Database session
        legal_entity_id: Legal entity ID
        instrument_type: Filter by instrument type (None = all types)
        with_terminated: Whether to include terminated instruments

    Returns:
        List of financial instruments
    """
    models = (
        FinancialInstrumentModelBroker.list_financial_instruments_for_legal_entity(
            session,
            legal_entity_id=legal_entity_id,
            instrument_type=instrument_type,  # type: ignore[arg-type]
            with_terminated=with_terminated,
        )
    )
    return [_to_financial_instrument(m) for m in models]

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

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

LegalEntityQueries

LegalEntityQueries()

Queries for legal entity operations.

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

def __init__(self) -> None:
    pass

create classmethod

create()

Normal factory.

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

@classmethod
def create(cls) -> "LegalEntityQueries":
    """Normal factory."""
    return cls()

create_null classmethod

create_null()

Null factory.

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

@classmethod
def create_null(cls) -> "LegalEntityQueries":
    """Null factory."""
    return cls()

get_legal_entity_provider_mapping

get_legal_entity_provider_mapping(
    session, /, legal_entity_id, workspace_key
)

Get provider mapping for a legal entity in a workspace.

Parameters:

Name	Type	Description	Default
session	Session	Database session	required
legal_entity_id	LegalEntityId	Legal entity ID	required
workspace_key	str	Provider workspace key	required

Returns:

Type	Description
LegalEntityProviderMapping | None	The provider mapping or None

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

def get_legal_entity_provider_mapping(
    self,
    session: Session,
    /,
    legal_entity_id: LegalEntityId,
    workspace_key: str,
) -> LegalEntityProviderMapping | None:
    """Get provider mapping for a legal entity in a workspace.

    Args:
        session: Database session
        legal_entity_id: Legal entity ID
        workspace_key: Provider workspace key

    Returns:
        The provider mapping or None
    """
    model = (
        LegalEntityProviderMappingModelBroker.get_provider_mapping_for_workspace(
            session,
            legal_entity_id=legal_entity_id,
            workspace_key=workspace_key,
        )
    )
    if model is None:
        return None
    return _to_legal_entity_provider_mapping(model)