pyeudiw.storage package

Submodules

pyeudiw.storage.base_cache module

class pyeudiw.storage.base_cache.BaseCache

Bases: BaseDB

Interface class for cache storage.

overwrite(object_name: str, value_gen_fn: Callable[[], str]) → dict

Overwrite an object in the cache.

Parameters:

• object_name (str) – the name of the object to overwrite.

• value_gen_fn (Callable[[], str]) – the function to call to generate the new value.

Returns:

the overwritten object.

Return type:

dict

set(data: dict) → dict

Set an object in the cache.

Parameters:

data (dict) – the data to set.

Returns:

the setted object.

Return type:

dict

try_retrieve(object_name: str, on_not_found: Callable[[], str]) → tuple[dict, RetrieveStatus]

Try to retrieve an object from the cache. If the object is not found, call the on_not_found function.

Parameters:

• object_name (str) – the name of the object to retrieve.

• on_not_found (Callable[[], str]) – the function to call if the object is not found.

Returns:

a tuple with the retrieved object and a status.

Return type:

tuple[dict, RetrieveStatus]

class pyeudiw.storage.base_cache.RetrieveStatus(*values)

Bases: Enum

ADDED = 1

RETRIEVED = 0

pyeudiw.storage.base_db module

class pyeudiw.storage.base_db.BaseDB

Bases: object

Interface class for database storage.

close() → None

Close the connection to the storage.

Returns:

None

pyeudiw.storage.base_storage module

class pyeudiw.storage.base_storage.BaseStorage

Bases: BaseDB

Interface class for storage.

add_dpop_proof_and_attestation(document_id, dpop_proof: dict, attestation: dict) → UpdateResult

Add a dpop proof and an attestation to the session.

Parameters:

• document_id (str) – the document id.

• dpop_proof (dict) – the dpop proof.

• attestation (dict) – the attestation.

Returns:

the result of the update operation.

Return type:

UpdateResult

add_empty_trust_anchor(entity_id: str) → str

Add an empty trust anchor.

Parameters:

entity_id (str) – the entity id.

Returns:

the document id.

Return type:

str

add_or_update_trust_attestation(entity_id: str, attestation: list[str], exp: <module 'datetime' from '/opt/hostedtoolcache/Python/3.12.12/x64/lib/python3.12/datetime.py'>) → str

Add or update a trust attestation.

Parameters:

• entity_id (str) – the entity id.

• attestation (list[str]) – the attestation.

• exp (datetime) – the expiration date.

Returns:

the document id.

Return type:

str

add_trust_anchor(entity_id: str, entity_configuration: str, exp: <module 'datetime' from '/opt/hostedtoolcache/Python/3.12.12/x64/lib/python3.12/datetime.py'>, trust_type: ~pyeudiw.storage.base_storage.TrustType)

Add a trust anchor.

Parameters:

• entity_id (str) – the entity id.

• entity_configuration (str) – the entity configuration.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

Returns:

the document id.

Return type:

str

add_trust_attestation(entity_id: str, attestation: list[str], exp: <module 'datetime' from '/opt/hostedtoolcache/Python/3.12.12/x64/lib/python3.12/datetime.py'>, trust_type: ~pyeudiw.storage.base_storage.TrustType, jwks: dict) → str

Add a trust attestation.

Parameters:

• entity_id (str) – the entity id.

• attestation (list[str]) – the attestation.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

• jwks (dict) – cached jwks

Returns:

the document id.

Return type:

str

add_trust_attestation_metadata(entity_id: str, metadata_type: str, metadata: dict) → str

Add a trust attestation metadata.

Parameters:

• entity_id (str) – the entity id.

• metadata_type (str) – the metadata type.

Returns:

the document id.

Return type:

str

add_trust_source(entity_id: str, trust_source: dict) → str

Add a trust source.

Parameters:

• entity_id (str) – the entity id.

• trust_source (dict) – the trust source.

Returns:

the document id.

Return type:

str

exists_by_state_and_session_id(state: str, session_id: str = '') → bool

Check if a session exists by state and session id.

Parameters:

• state (str) – the state.

• session_id (str) – the session id.

Returns:

True if the session exists, False otherwise.

Return type:

bool

get_by_nonce_state(state: str, nonce: str) → dict | None

Get a session by nonce and state.

Parameters:

• state (str) – the state.

• nonce (str) – the nonce.

Returns:

the session.

Return type:

Union[dict, None]

get_by_session_id(session_id: str) → dict | None

Get a session by session id.

Parameters:

session_id (str) – the session id.

Returns:

the session.

Return type:

Union[dict, None]

get_by_state(state: str) → dict | None

Get a session by state.

Parameters:

state (str) – the state.

Returns:

the session.

Return type:

Union[dict, None]

get_by_state_and_session_id(state: str, session_id: str = '') → dict | None

Get a session by state and session id.

Parameters:

• state (str) – the state.

• session_id (str) – the session id.

Returns:

the session.

Return type:

Union[dict, None]

get_trust_anchor(entity_id: str) → dict | None

Get a trust anchor.

Parameters:

entity_id (str) – the entity id.

Returns:

the trust anchor.

Return type:

Union[dict, None]

get_trust_attestation(entity_id: str) → dict | None

Get a trust attestation.

Parameters:

entity_id (str) – the entity id.

Returns:

the trust attestation.

Return type:

Union[dict, None]

get_trust_source(entity_id: str) → dict | None

Get a trust source.

Parameters:

entity_id (str) – the entity id.

Returns:

the trust source.

Return type:

Union[dict, None]

has_session_retention_ttl() → bool

Check if the session has a retention ttl.

Returns:

True if the session has a retention ttl, False otherwise.

Return type:

bool

has_trust_anchor(entity_id: str) → bool

Check if a trust anchor exists.

Parameters:

entity_id (str) – the entity id.

Returns:

True if the trust anchor exists, False otherwise.

Return type:

bool

has_trust_attestation(entity_id: str) → bool

Check if a trust attestation exists.

Parameters:

entity_id (str) – the entity id.

Returns:

True if the trust attestation exists, False otherwise.

Return type:

bool

has_trust_source(entity_id: str) → bool

init_session(document_id: str, session_id: str, state: str, remote_flow_typ: str) → str

Initialize a session.

Parameters:

• document_id (str) – the document id.

• session_id – the satosa session id that initiated the authentication flow.

• state (str) – a unique identifier of the authentication flow

• remote_flow_typ (str) – a value that discriminates between different authentication flow

property is_connected: bool

Check if the storage is connected.

Returns:

True if the storage is connected, False otherwise.

Return type:

bool

search_session_by_field(field: str, value: str) → dict

Search for a session by a specific field and value.

Parameters:

• field (str) – the field to search by.

• value (str) – the value to search for.

Returns:

the session data if found, otherwise an empty dict.

Return type:

dict

set_finalized(document_id: str) → UpdateResult

Set the session as finalized.

Parameters:

document_id (str) – the document id.

Returns:

the result of the update operation.

Return type:

UpdateResult

set_session_retention_ttl(ttl: int) → None

Set the database retention ttl.

Parameters:

ttl (int | None) – the ttl.

update_request_object(document_id: str, request_object: dict) → UpdateResult

Update the request object of the session.

Parameters:

• document_id (str) – the document id.

• request_object (dict) – the request object.

Returns:

the result of the update operation.

Return type:

UpdateResult

update_response_object(nonce: str, state: str, response_object: dict, isError: bool = False) → UpdateResult

Update the response object of the session.

Parameters:

• nonce (str) – the nonce.

• state (str) – the state.

• response_object (dict) – the response object.

• isError (bool) – if the response is an error response.

Returns:

the result of the update operation.

Return type:

UpdateResult

update_trust_anchor(entity_id: str, entity_configuration: str, exp: <module 'datetime' from '/opt/hostedtoolcache/Python/3.12.12/x64/lib/python3.12/datetime.py'>, trust_type: ~pyeudiw.storage.base_storage.TrustType) → str

Update a trust anchor.

Parameters:

• entity_id (str) – the entity id.

• entity_configuration (str) – the entity configuration.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

Returns:

the document id.

Return type:

str

update_trust_attestation(entity_id: str, attestation: list[str], exp: <module 'datetime' from '/opt/hostedtoolcache/Python/3.12.12/x64/lib/python3.12/datetime.py'>, trust_type: ~pyeudiw.storage.base_storage.TrustType, jwks: dict) → str

Update a trust attestation.

Parameters:

• entity_id (str) – the entity id.

• attestation (list[str]) – the attestation.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

• jwks (dict) – cached jwks

Returns:

the document id.

Return type:

str

upsert_session(session_id: str, data: dict) → tuple[str, dict]

Upsert a session by session id.

Parameters:

• session_id (str) – the session id.

• data (dict) – the data to upsert.

Returns:

a tuple containing the document id and the updated data.

Return type:

tuple[str, dict]

class pyeudiw.storage.base_storage.TrustType(*values)

Bases: Enum

DIRECT_TRUST_SD_JWT_VC = 'direct_trust_sd_jwt_vc'

FEDERATION = 'federation'

X509 = 'x509'

pyeudiw.storage.credential_entity module

class pyeudiw.storage.credential_entity.CredentialEntity(*, user_id: str, incremental_id: int = <factory>, revoked: bool = False, identifier: str)

Bases: BaseModel

Data model representing a user credential entity for credential issuance.

identifier: str

incremental_id: int

model_config: ClassVar[ConfigDict] = {}

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

revoked: bool

user_id: str

pyeudiw.storage.credential_storage module

class pyeudiw.storage.credential_storage.CredentialStorage(conf: dict, url: str, connection_params=None)

Bases: MongoStorage

A storage class extending MongoStorage to manage user credentials for OpenID4VCI interactions.

This class provides methods to initialize, retrieve, and update session data stored in a MongoDB database.

close()

Close the connection to the storage.

Returns:

None

get_all_sorted_by_incremental_id(sort_direction=1) → list[dict]

get_by_field(field_name: str, field_value: str) → CredentialEntity

get_by_fields(query: dict) → CredentialEntity

get_credential_by_user_id(user_id: str) → CredentialEntity

property is_connected: bool

Check if the storage is connected.

Returns:

True if the storage is connected, False otherwise.

Return type:

bool

set_session_retention_ttl(ttl: int) → None

Set the database retention ttl.

Parameters:

ttl (int | None) – the ttl.

pyeudiw.storage.db_engine module

class pyeudiw.storage.db_engine.DBEngine(config: dict)

Bases: BaseStorage, BaseCache, BaseLogger

DB Engine class.

add_dpop_proof_and_attestation(document_id, dpop_proof: dict, attestation: dict)

Add a dpop proof and an attestation to the session.

Parameters:

• document_id (str) – the document id.

• dpop_proof (dict) – the dpop proof.

• attestation (dict) – the attestation.

Returns:

the result of the update operation.

Return type:

UpdateResult

add_empty_trust_anchor(entity_id: str) → str

Add an empty trust anchor.

Parameters:

entity_id (str) – the entity id.

Returns:

the document id.

Return type:

str

add_or_update_trust_attestation(entity_id: str, attestation: list[str] = [], exp: datetime = None, trust_type: TrustType = TrustType.FEDERATION, jwks: list[dict] = []) → str

Add or update a trust attestation.

Parameters:

• entity_id (str) – the entity id.

• attestation (list[str]) – the attestation.

• exp (datetime) – the expiration date.

Returns:

the document id.

Return type:

str

add_trust_anchor(entity_id: str, entity_configuration: str, exp: datetime, trust_type: TrustType = TrustType.FEDERATION) → str

Add a trust anchor.

Parameters:

• entity_id (str) – the entity id.

• entity_configuration (str) – the entity configuration.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

Returns:

the document id.

Return type:

str

add_trust_attestation(entity_id: str, attestation: list[str] = [], exp: datetime = None, trust_type: TrustType = TrustType.FEDERATION, jwks: list[dict] = []) → str

Add a trust attestation.

Parameters:

• entity_id (str) – the entity id.

• attestation (list[str]) – the attestation.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

• jwks (dict) – cached jwks

Returns:

the document id.

Return type:

str

add_trust_attestation_metadata(entity_id: str, metadat_type: str, metadata: dict) → str

Add a trust attestation metadata.

Parameters:

• entity_id (str) – the entity id.

• metadata_type (str) – the metadata type.

Returns:

the document id.

Return type:

str

add_trust_source(trust_source: dict) → str

Add a trust source.

Parameters:

• entity_id (str) – the entity id.

• trust_source (dict) – the trust source.

Returns:

the document id.

Return type:

str

close()

Close the connection to the storage.

Returns:

None

exists_by_state_and_session_id(state: str, session_id: str = '') → bool

Check if a session exists by state and session id.

Parameters:

• state (str) – the state.

• session_id (str) – the session id.

Returns:

True if the session exists, False otherwise.

Return type:

bool

get(method: str, *args, **kwargs) → dict | None

Perform a get operation on the storages.

Parameters:

• method (str) – the method to call.

• args (Any) – the arguments to pass to the method.

• kwargs (Any) – the keyword arguments to pass to the method.

Raises:

EntryNotFound – if the entry is not found on any storage.

Returns:

the result of the first elment found on DBs.

Return type:

Union[dict, None]

get_by_nonce_state(state: str, nonce: str) → dict | None

Get a session by nonce and state.

Parameters:

• state (str) – the state.

• nonce (str) – the nonce.

Returns:

the session.

Return type:

Union[dict, None]

get_by_session_id(session_id: str) → dict | None

Get a session by session id.

Parameters:

session_id (str) – the session id.

Returns:

the session.

Return type:

Union[dict, None]

get_by_state(state: str) → dict | None

Get a session by state.

Parameters:

state (str) – the state.

Returns:

the session.

Return type:

Union[dict, None]

get_by_state_and_session_id(state: str, session_id: str = '') → dict | None

Get a session by state and session id.

Parameters:

• state (str) – the state.

• session_id (str) – the session id.

Returns:

the session.

Return type:

Union[dict, None]

get_trust_anchor(entity_id: str) → dict | None

Get a trust anchor.

Parameters:

entity_id (str) – the entity id.

Returns:

the trust anchor.

Return type:

Union[dict, None]

get_trust_attestation(entity_id: str) → dict | None

Get a trust attestation.

Parameters:

entity_id (str) – the entity id.

Returns:

the trust attestation.

Return type:

Union[dict, None]

get_trust_source(entity_id: str) → dict

Get a trust source.

Parameters:

entity_id (str) – the entity id.

Returns:

the trust source.

Return type:

Union[dict, None]

has_trust_anchor(entity_id: str) → bool

Check if a trust anchor exists.

Parameters:

entity_id (str) – the entity id.

Returns:

True if the trust anchor exists, False otherwise.

Return type:

bool

has_trust_attestation(entity_id: str) → bool

Check if a trust attestation exists.

Parameters:

entity_id (str) – the entity id.

Returns:

True if the trust attestation exists, False otherwise.

Return type:

bool

has_trust_source(entity_id: str) → bool

init_session(session_id: str, state: str, remote_flow_typ: str) → str

Initialize a session.

Parameters:

• document_id (str) – the document id.

• session_id – the satosa session id that initiated the authentication flow.

• state (str) – a unique identifier of the authentication flow

• remote_flow_typ (str) – a value that discriminates between different authentication flow

property is_connected

Check if the storage is connected.

Returns:

True if the storage is connected, False otherwise.

Return type:

bool

overwrite(object_name: str, value_gen_fn: Callable[[], str]) → dict

Overwrite an object in the cache.

Parameters:

• object_name (str) – the name of the object to overwrite.

• value_gen_fn (Callable[[], str]) – the function to call to generate the new value.

Returns:

the overwritten object.

Return type:

dict

search_session_by_field(field: str, value: str) → dict | None

Search for a session by a specific field and value.

Parameters:

• field (str) – the field to search by.

• value (str) – the value to search for.

Returns:

the session data if found, otherwise an empty dict.

Return type:

dict

set_finalized(document_id: str)

Set the session as finalized.

Parameters:

document_id (str) – the document id.

Returns:

the result of the update operation.

Return type:

UpdateResult

try_retrieve(object_name: str, on_not_found: Callable[[], str]) → dict

Try to retrieve an object from the cache. If the object is not found, call the on_not_found function.

Parameters:

• object_name (str) – the name of the object to retrieve.

• on_not_found (Callable[[], str]) – the function to call if the object is not found.

Returns:

a tuple with the retrieved object and a status.

Return type:

tuple[dict, RetrieveStatus]

update_request_object(document_id: str, request_object: dict) → int

Update the request object of the session.

Parameters:

• document_id (str) – the document id.

• request_object (dict) – the request object.

Returns:

the result of the update operation.

Return type:

UpdateResult

update_response_object(nonce: str, state: str, response_object: dict, isError: bool = False) → int

Update the response object of the session.

Parameters:

• nonce (str) – the nonce.

• state (str) – the state.

• response_object (dict) – the response object.

• isError (bool) – if the response is an error response.

Returns:

the result of the update operation.

Return type:

UpdateResult

update_trust_anchor(entity_id: str, entity_configuration: dict, exp: datetime, trust_type: TrustType = TrustType.FEDERATION) → str

Update a trust anchor.

Parameters:

• entity_id (str) – the entity id.

• entity_configuration (str) – the entity configuration.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

Returns:

the document id.

Return type:

str

update_trust_attestation(entity_id: str, attestation: list[str] = [], exp: datetime = None, trust_type: TrustType = TrustType.FEDERATION, jwks: list[dict] = []) → str

Update a trust attestation.

Parameters:

• entity_id (str) – the entity id.

• attestation (list[str]) – the attestation.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

• jwks (dict) – cached jwks

Returns:

the document id.

Return type:

str

upsert_session(session_id: str, data: dict) → int

Upsert a session by session id.

Parameters:

• session_id (str) – the session id.

• data (dict) – the data to upsert.

Returns:

a tuple containing the document id and the updated data.

Return type:

tuple[str, dict]

write(method: str, *args, **kwargs)

Perform a write operation on the storages.

Parameters:

• method (str) – the method to call.

• args (Any) – the arguments to pass to the method.

• kwargs (Any) – the keyword arguments to pass to the method.

Raises:

StorageWriteError – if the write operation fails on all the storages.

Returns:

the number of replicas where the write operation is successful.

Return type:

int

pyeudiw.storage.exceptions module

exception pyeudiw.storage.exceptions.ChainNotExist

Bases: Exception

exception pyeudiw.storage.exceptions.EntryNotFound

Bases: Exception

exception pyeudiw.storage.exceptions.StorageEntryUpdateFailed

Bases: Exception

exception pyeudiw.storage.exceptions.StorageWriteError

Bases: Exception

pyeudiw.storage.mongo_cache module

class pyeudiw.storage.mongo_cache.MongoCache(conf: dict, url: str, connection_params: dict = None)

Bases: BaseCache

MongoDB cache implementation.

close() → None

Close the connection to the storage.

Returns:

None

overwrite(object_name: str, value_gen_fn: Callable[[], str]) → dict

Overwrite an object in the cache.

Parameters:

• object_name (str) – the name of the object to overwrite.

• value_gen_fn (Callable[[], str]) – the function to call to generate the new value.

Returns:

the overwritten object.

Return type:

dict

set(data: dict) → dict

Set an object in the cache.

Parameters:

data (dict) – the data to set.

Returns:

the setted object.

Return type:

dict

try_retrieve(object_name: str, on_not_found: Callable[[], str]) → tuple[dict, RetrieveStatus]

Try to retrieve an object from the cache. If the object is not found, call the on_not_found function.

Parameters:

• object_name (str) – the name of the object to retrieve.

• on_not_found (Callable[[], str]) – the function to call if the object is not found.

Returns:

a tuple with the retrieved object and a status.

Return type:

tuple[dict, RetrieveStatus]

pyeudiw.storage.mongo_storage module

class pyeudiw.storage.mongo_storage.MongoStorage(conf: dict, url: str, connection_params: dict = {})

Bases: BaseStorage

add_dpop_proof_and_attestation(document_id: str, dpop_proof: dict, attestation: dict) → UpdateResult

Add a dpop proof and an attestation to the session.

Parameters:

• document_id (str) – the document id.

• dpop_proof (dict) – the dpop proof.

• attestation (dict) – the attestation.

Returns:

the result of the update operation.

Return type:

UpdateResult

add_empty_trust_anchor(entity_id: str) → str

Add an empty trust anchor.

Parameters:

entity_id (str) – the entity id.

Returns:

the document id.

Return type:

str

add_trust_anchor(entity_id: str, entity_configuration: str, exp: datetime, trust_type: TrustType)

Add a trust anchor.

Parameters:

• entity_id (str) – the entity id.

• entity_configuration (str) – the entity configuration.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

Returns:

the document id.

Return type:

str

add_trust_attestation(entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType, jwks: list[dict]) → str

Add a trust attestation.

Parameters:

• entity_id (str) – the entity id.

• attestation (list[str]) – the attestation.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

• jwks (dict) – cached jwks

Returns:

the document id.

Return type:

str

add_trust_attestation_metadata(entity_id: str, metadata_type: str, metadata: dict)

Add a trust attestation metadata.

Parameters:

• entity_id (str) – the entity id.

• metadata_type (str) – the metadata type.

Returns:

the document id.

Return type:

str

add_trust_source(trust_source: dict) → str

Add a trust source.

Parameters:

• entity_id (str) – the entity id.

• trust_source (dict) – the trust source.

Returns:

the document id.

Return type:

str

close()

Close the connection to the storage.

Returns:

None

get_by_id(document_id: str) → dict

get_by_nonce_state(nonce: str, state: str | None) → dict

Get a session by nonce and state.

Parameters:

• state (str) – the state.

• nonce (str) – the nonce.

Returns:

the session.

Return type:

Union[dict, None]

get_by_session_id(session_id: str) → dict | None

Get a session by session id.

Parameters:

session_id (str) – the session id.

Returns:

the session.

Return type:

Union[dict, None]

get_by_state_and_session_id(state: str, session_id: str = '') → dict | None

Get a session by state and session id.

Parameters:

• state (str) – the state.

• session_id (str) – the session id.

Returns:

the session.

Return type:

Union[dict, None]

get_session_retention_ttl() → dict

get_trust_anchor(entity_id: str) → dict | None

Get a trust anchor.

Parameters:

entity_id (str) – the entity id.

Returns:

the trust anchor.

Return type:

Union[dict, None]

get_trust_attestation(entity_id: str) → dict | None

Get a trust attestation.

Parameters:

entity_id (str) – the entity id.

Returns:

the trust attestation.

Return type:

Union[dict, None]

get_trust_source(entity_id: str) → dict | None

Get a trust source.

Parameters:

entity_id (str) – the entity id.

Returns:

the trust source.

Return type:

Union[dict, None]

has_session_retention_ttl() → bool

Check if the session has a retention ttl.

Returns:

True if the session has a retention ttl, False otherwise.

Return type:

bool

has_trust_anchor(entity_id: str) → bool

Check if a trust anchor exists.

Parameters:

entity_id (str) – the entity id.

Returns:

True if the trust anchor exists, False otherwise.

Return type:

bool

has_trust_attestation(entity_id: str) → bool

Check if a trust attestation exists.

Parameters:

entity_id (str) – the entity id.

Returns:

True if the trust attestation exists, False otherwise.

Return type:

bool

has_trust_source(entity_id: str) → bool

init_session(document_id: str, session_id: str, state: str, remote_flow_typ: str) → str

Initialize a session.

Parameters:

• document_id (str) – the document id.

• session_id – the satosa session id that initiated the authentication flow.

• state (str) – a unique identifier of the authentication flow

• remote_flow_typ (str) – a value that discriminates between different authentication flow

property is_connected: bool

Check if the storage is connected.

Returns:

True if the storage is connected, False otherwise.

Return type:

bool

search_session_by_field(field: str, value: str) → dict

Search for a session by a specific field and value.

Parameters:

• field (str) – the field to search by.

• value (str) – the value to search for.

Returns:

the session data if found, otherwise an empty dict.

Return type:

dict

set_finalized(document_id: str)

Set the session as finalized.

Parameters:

document_id (str) – the document id.

Returns:

the result of the update operation.

Return type:

UpdateResult

set_session_retention_ttl(ttl: int) → None

Set the database retention ttl.

Parameters:

ttl (int | None) – the ttl.

update_request_object(document_id: str, request_object: dict) → UpdateResult

Update the request object of the session.

Parameters:

• document_id (str) – the document id.

• request_object (dict) – the request object.

Returns:

the result of the update operation.

Return type:

UpdateResult

update_response_object(nonce: str, state: str, internal_response: dict, isError: bool = False) → UpdateResult

Update the response object of the session.

Parameters:

• nonce (str) – the nonce.

• state (str) – the state.

• response_object (dict) – the response object.

• isError (bool) – if the response is an error response.

Returns:

the result of the update operation.

Return type:

UpdateResult

update_trust_anchor(entity_id: str, entity_configuration: str, exp: datetime, trust_type: TrustType) → str

Update a trust anchor.

Parameters:

• entity_id (str) – the entity id.

• entity_configuration (str) – the entity configuration.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

Returns:

the document id.

Return type:

str

update_trust_attestation(entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType, jwks: list[dict]) → str

Update a trust attestation.

Parameters:

• entity_id (str) – the entity id.

• attestation (list[str]) – the attestation.

• exp (datetime) – the expiration date.

• trust_type (TrustType) – the trust type.

• jwks (dict) – cached jwks

Returns:

the document id.

Return type:

str

upsert_session(session_id: str, data: dict) → tuple[str, dict]

Upsert a session by session id.

Parameters:

• session_id (str) – the session id.

• data (dict) – the data to upsert.

Returns:

a tuple containing the document id and the updated data.

Return type:

tuple[str, dict]

pyeudiw.storage.user_credential_db_engine module

class pyeudiw.storage.user_credential_db_engine.UserCredentialEngine(config: dict)

Bases: object

Engine for managing User and Credential storage operations.

This class provides a wrapper around the configured storage backend, typically a MongoDB-based engine, used for persisting and retrieving user and credential related data. It lazily initializes the DB engine and ensures it is connected when accessed.

property db_credential_storage_engine: CredentialStorage

Lazily initialized access to MongoDB storage engine.

Returns:

CredentialStorage: The initialized DB engine instance.

property db_user_storage_engine: UserStorage

Lazily initialized access to MongoDB storage engine.

Returns:

UserStorage: The initialized DB engine instance.

pyeudiw.storage.user_entity module

class pyeudiw.storage.user_entity.UserEntity(*, name: str, surname: str, dateOfBirth: str, fiscal_code: str, countyOfBirth: str, placeOfBirth: str, portrait: str | None = None, mail: str)

Bases: BaseModel

Data model representing a user entity for credential issuance.

This model contains the core identity attributes typically included in digital credentials, such as personal details and place of birth.

Attributes:

name (str): The user’s given (first) name. surname (str): The user’s family (last) name. dateOfBirth (str): The user’s date of birth in ISO format (YYYY-MM-DD). fiscal_code (str): The user’s unique personal identifier, such as a fiscal code. countyOfBirth (str): The ISO country code of the user’s place of birth (e.g., “IT”). placeOfBirth (str): The locality (city, town) where the user was born. portrait (Optional[str]): An optional base64-encoded image of the user’s portrait.

countyOfBirth: str

dateOfBirth: str

fiscal_code: str

mail: str

model_config: ClassVar[ConfigDict] = {}

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

name: str

placeOfBirth: str

portrait: str | None

surname: str

pyeudiw.storage.user_storage module

class pyeudiw.storage.user_storage.UserStorage(conf: dict, url: str, connection_params=None)

Bases: MongoStorage

A storage class extending MongoStorage to manage user for OpenID4VCI interactions.

This class provides methods to initialize, retrieve, and update session data stored in a MongoDB database.

close()

Close the connection to the storage.

Returns:

None

get_by_field(field_name: str, field_value: str) → tuple[str, UserEntity]

get_by_fields(query: dict) → tuple[str, UserEntity]

get_by_fiscal_code(fiscal_code: str) → tuple[str, UserEntity]

property is_connected: bool

Check if the storage is connected.

Returns:

True if the storage is connected, False otherwise.

Return type:

bool

set_session_retention_ttl(ttl: int) → None

Set the database retention ttl.

Parameters:

ttl (int | None) – the ttl.

upsert_user(user_entity: UserEntity | dict) → str

Module contents