Keycloak

The keycloak adapter provides integration with Keycloak for identity and access management, including token validation, user management, and role-based access control.

Ports

Abstract port interface defining the Keycloak adapter contract.

Keycloak port definitions for ArchiPy.

archipy.adapters.keycloak.ports.KeycloakResponseType module-attribute

KeycloakResponseType = dict[str, Any]

archipy.adapters.keycloak.ports.KeycloakRoleType module-attribute

KeycloakRoleType = dict[str, Any]

archipy.adapters.keycloak.ports.KeycloakUserType module-attribute

KeycloakUserType = dict[str, Any]

archipy.adapters.keycloak.ports.KeycloakGroupType module-attribute

KeycloakGroupType = dict[str, Any]

archipy.adapters.keycloak.ports.KeycloakTokenType module-attribute

KeycloakTokenType = dict[str, Any]

archipy.adapters.keycloak.ports.KeycloakOrganizationType module-attribute

KeycloakOrganizationType = dict[str, Any]

archipy.adapters.keycloak.ports.PublicKeyType module-attribute

PublicKeyType = Any

archipy.adapters.keycloak.ports.KeycloakPort

Interface for Keycloak operations providing a standardized access pattern.

This interface defines the contract for Keycloak adapters, ensuring consistent implementation of Keycloak operations across different adapters. It covers essential functionality including authentication, user management, and role management.

Source code in archipy/adapters/keycloak/ports.py

class KeycloakPort:
    """Interface for Keycloak operations providing a standardized access pattern.

    This interface defines the contract for Keycloak adapters, ensuring consistent
    implementation of Keycloak operations across different adapters. It covers essential
    functionality including authentication, user management, and role management.
    """

    # Token Operations
    @abstractmethod
    def get_token(self, username: str, password: str) -> KeycloakTokenType | None:
        """Get a user token by username and password."""
        raise NotImplementedError

    @abstractmethod
    def refresh_token(self, refresh_token: str) -> KeycloakTokenType | None:
        """Refresh an existing token using a refresh token."""
        raise NotImplementedError

    @abstractmethod
    def validate_token(self, token: str) -> bool:
        """Validate if a token is still valid."""
        raise NotImplementedError

    @abstractmethod
    def get_userinfo(self, token: str) -> KeycloakUserType | None:
        """Get user information from a token."""
        raise NotImplementedError

    @abstractmethod
    def get_token_info(self, token: str) -> dict[str, Any] | None:
        """Decode token to get its claims."""
        raise NotImplementedError

    @abstractmethod
    def introspect_token(self, token: str) -> dict[str, Any] | None:
        """Introspect token to get detailed information about it."""
        raise NotImplementedError

    @abstractmethod
    def get_client_credentials_token(self) -> KeycloakTokenType | None:
        """Get token using client credentials."""
        raise NotImplementedError

    @abstractmethod
    def logout(self, refresh_token: str) -> None:
        """Logout user by invalidating their refresh token."""
        raise NotImplementedError

    # User Operations
    @abstractmethod
    def get_user_by_id(self, user_id: str) -> KeycloakUserType | None:
        """Get user details by user ID."""
        raise NotImplementedError

    @abstractmethod
    def get_user_by_username(self, username: str) -> KeycloakUserType | None:
        """Get user details by username."""
        raise NotImplementedError

    @abstractmethod
    def get_user_by_email(self, email: str) -> KeycloakUserType | None:
        """Get user details by email."""
        raise NotImplementedError

    @abstractmethod
    def create_user(self, user_data: dict[str, Any]) -> str | None:
        """Create a new user in Keycloak."""
        raise NotImplementedError

    @abstractmethod
    def update_user(self, user_id: str, user_data: dict[str, Any]) -> None:
        """Update user details."""
        raise NotImplementedError

    @abstractmethod
    def reset_password(self, user_id: str, password: str, temporary: bool = False) -> None:
        """Reset a user's password."""
        raise NotImplementedError

    @abstractmethod
    def search_users(self, query: str, max_results: int = 100) -> list[KeycloakUserType]:
        """Search for users by username, email, or name."""
        raise NotImplementedError

    @abstractmethod
    def clear_user_sessions(self, user_id: str) -> None:
        """Clear all sessions for a user."""
        raise NotImplementedError

    # Role Operations
    @abstractmethod
    def get_user_roles(self, user_id: str) -> list[KeycloakRoleType]:
        """Get roles assigned to a user."""
        raise NotImplementedError

    @abstractmethod
    def get_client_roles_for_user(self, user_id: str, client_id: str) -> list[KeycloakRoleType]:
        """Get client-specific roles assigned to a user."""
        raise NotImplementedError

    @abstractmethod
    def has_role(self, token: str, role_name: str) -> bool:
        """Check if a user has a specific role."""
        raise NotImplementedError

    @abstractmethod
    def has_any_of_roles(self, token: str, role_names: frozenset[str]) -> bool:
        """Check if a user has any of the specified roles."""
        raise NotImplementedError

    @abstractmethod
    def has_all_roles(self, token: str, role_names: frozenset[str]) -> bool:
        """Check if a user has all of the specified roles."""
        raise NotImplementedError

    @abstractmethod
    def assign_realm_role(self, user_id: str, role_name: str) -> None:
        """Assign a realm role to a user."""
        raise NotImplementedError

    @abstractmethod
    def remove_realm_role(self, user_id: str, role_name: str) -> None:
        """Remove a realm role from a user."""
        raise NotImplementedError

    @abstractmethod
    def assign_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
        """Assign a client-specific role to a user."""
        raise NotImplementedError

    @abstractmethod
    def remove_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
        """Remove a client-specific role from a user."""
        raise NotImplementedError

    @abstractmethod
    def get_realm_role(self, role_name: str) -> dict[str, Any]:
        """Get realm role."""
        raise NotImplementedError

    @abstractmethod
    def get_realm_roles(self) -> list[dict[str, Any]]:
        """Get all realm roles."""
        raise NotImplementedError

    @abstractmethod
    def create_realm_role(
        self,
        role_name: str,
        description: str | None = None,
        skip_exists: bool = True,
    ) -> dict[str, Any] | None:
        """Create a new realm role."""
        raise NotImplementedError

    @abstractmethod
    def delete_realm_role(self, role_name: str) -> None:
        """Delete a realm role."""
        raise NotImplementedError

    # Client Operations
    @abstractmethod
    def get_client_id(self, client_name: str) -> str:
        """Get client ID by client name."""
        raise NotImplementedError

    @abstractmethod
    def get_client_secret(self, client_id: str) -> str:
        """Get client secret."""
        raise NotImplementedError

    @abstractmethod
    def get_service_account_id(self) -> str:
        """Get service account user ID for the current client."""
        raise NotImplementedError

    # System Operations
    @abstractmethod
    def get_public_key(self) -> PublicKeyType:
        """Get the public key used to verify tokens."""
        raise NotImplementedError

    @abstractmethod
    def get_well_known_config(self) -> dict[str, Any]:
        """Get the well-known OpenID configuration."""
        raise NotImplementedError

    @abstractmethod
    def get_certs(self) -> dict[str, Any]:
        """Get the JWT verification certificates."""
        raise NotImplementedError

    # Authorization
    @abstractmethod
    def get_token_from_code(self, code: str, redirect_uri: str) -> KeycloakTokenType | None:
        """Exchange authorization code for token."""
        raise NotImplementedError

    @abstractmethod
    def check_permissions(self, token: str, resource: str, scope: str) -> bool:
        """Check if a user has permission to access a resource with the specified scope."""
        raise NotImplementedError

    @abstractmethod
    def delete_user(self, user_id: str) -> None:
        """Delete a user from Keycloak by their ID."""
        raise NotImplementedError

    @abstractmethod
    def create_client_role(
        self,
        client_id: str,
        role_name: str,
        description: str | None = None,
        skip_exists: bool = True,
    ) -> dict[str, Any] | None:
        """Create a new client role."""
        raise NotImplementedError

    @abstractmethod
    def create_realm(self, realm_name: str, skip_exists: bool = True, **kwargs: Any) -> dict[str, Any] | None:
        """Create a new Keycloak realm."""
        raise NotImplementedError

    @abstractmethod
    def get_realm(self, realm_name: str) -> dict[str, Any] | None:
        """Get realm details by realm name."""
        raise NotImplementedError

    @abstractmethod
    def update_realm(self, realm_name: str, **kwargs: Any) -> dict[str, Any] | None:
        """Update a realm. Kwargs are RealmRepresentation top-level attributes (e.g. displayName, organizationsEnabled)."""
        raise NotImplementedError

    @abstractmethod
    def create_client(
        self,
        client_id: str,
        realm: str | None = None,
        skip_exists: bool = True,
        **kwargs: Any,
    ) -> dict[str, Any] | None:
        """Create a new client in the specified realm."""
        raise NotImplementedError

    @abstractmethod
    def add_realm_roles_to_composite(self, composite_role_name: str, child_role_names: list[str]) -> None:
        """Add realm roles to a composite role."""
        raise NotImplementedError

    @abstractmethod
    def add_client_roles_to_composite(
        self,
        composite_role_name: str,
        client_id: str,
        child_role_names: list[str],
    ) -> None:
        """Add client roles to a composite role."""
        raise NotImplementedError

    @abstractmethod
    def get_composite_realm_roles(self, role_name: str) -> list[dict[str, Any]] | None:
        """Get composite roles for a realm role."""
        raise NotImplementedError

    # Organization Operations
    @abstractmethod
    def get_organizations(self, query: dict | None = None) -> list[KeycloakOrganizationType]:
        """Fetch all organizations. Returns list of OrganizationRepresentation, filtered by query."""
        raise NotImplementedError

    @abstractmethod
    def get_organization(self, organization_id: str) -> KeycloakOrganizationType:
        """Get representation of the organization by ID."""
        raise NotImplementedError

    @abstractmethod
    def create_organization(self, name: str, alias: str, **kwargs: Any) -> str | None:
        """Create a new organization. Name and alias must be unique. Returns org_id."""
        raise NotImplementedError

    @abstractmethod
    def update_organization(self, organization_id: str, **kwargs: Any) -> dict[str, Any]:
        """Update an existing organization. Kwargs are organization attributes (e.g. name, alias)."""
        raise NotImplementedError

    @abstractmethod
    def delete_organization(self, organization_id: str) -> dict[str, Any]:
        """Delete an organization."""
        raise NotImplementedError

    @abstractmethod
    def get_organization_idps(self, organization_id: str) -> list[dict[str, Any]]:
        """Get IDPs by organization id."""
        raise NotImplementedError

    @abstractmethod
    def get_user_organizations(self, user_id: str) -> list[KeycloakOrganizationType]:
        """Get organizations by user id. Returns list of organizations the user is member of."""
        raise NotImplementedError

    @abstractmethod
    def get_organization_members(self, organization_id: str, query: dict | None = None) -> list[dict[str, Any]]:
        """Get members by organization id, optionally filtered by query parameters."""
        raise NotImplementedError

    @abstractmethod
    def get_organization_members_count(self, organization_id: str) -> int:
        """Get the number of members in the organization."""
        raise NotImplementedError

    @abstractmethod
    def organization_user_add(self, user_id: str, organization_id: str) -> bytes:
        """Add a user to an organization."""
        raise NotImplementedError

    @abstractmethod
    def organization_user_remove(self, user_id: str, organization_id: str) -> dict[str, Any]:
        """Remove a user from an organization."""
        raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_token abstractmethod

get_token(
    username: str, password: str
) -> KeycloakTokenType | None

Get a user token by username and password.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_token(self, username: str, password: str) -> KeycloakTokenType | None:
    """Get a user token by username and password."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.refresh_token abstractmethod

refresh_token(
    refresh_token: str,
) -> KeycloakTokenType | None

Refresh an existing token using a refresh token.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def refresh_token(self, refresh_token: str) -> KeycloakTokenType | None:
    """Refresh an existing token using a refresh token."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.validate_token abstractmethod

validate_token(token: str) -> bool

Validate if a token is still valid.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def validate_token(self, token: str) -> bool:
    """Validate if a token is still valid."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_userinfo abstractmethod

get_userinfo(token: str) -> KeycloakUserType | None

Get user information from a token.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_userinfo(self, token: str) -> KeycloakUserType | None:
    """Get user information from a token."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_token_info abstractmethod

get_token_info(token: str) -> dict[str, Any] | None

Decode token to get its claims.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_token_info(self, token: str) -> dict[str, Any] | None:
    """Decode token to get its claims."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.introspect_token abstractmethod

introspect_token(token: str) -> dict[str, Any] | None

Introspect token to get detailed information about it.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def introspect_token(self, token: str) -> dict[str, Any] | None:
    """Introspect token to get detailed information about it."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_client_credentials_token abstractmethod

get_client_credentials_token() -> KeycloakTokenType | None

Get token using client credentials.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_client_credentials_token(self) -> KeycloakTokenType | None:
    """Get token using client credentials."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.logout abstractmethod

logout(refresh_token: str) -> None

Logout user by invalidating their refresh token.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def logout(self, refresh_token: str) -> None:
    """Logout user by invalidating their refresh token."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_user_by_id abstractmethod

get_user_by_id(user_id: str) -> KeycloakUserType | None

Get user details by user ID.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_user_by_id(self, user_id: str) -> KeycloakUserType | None:
    """Get user details by user ID."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_user_by_username abstractmethod

get_user_by_username(
    username: str,
) -> KeycloakUserType | None

Get user details by username.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_user_by_username(self, username: str) -> KeycloakUserType | None:
    """Get user details by username."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_user_by_email abstractmethod

get_user_by_email(email: str) -> KeycloakUserType | None

Get user details by email.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_user_by_email(self, email: str) -> KeycloakUserType | None:
    """Get user details by email."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.create_user abstractmethod

create_user(user_data: dict[str, Any]) -> str | None

Create a new user in Keycloak.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def create_user(self, user_data: dict[str, Any]) -> str | None:
    """Create a new user in Keycloak."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.update_user abstractmethod

update_user(
    user_id: str, user_data: dict[str, Any]
) -> None

Update user details.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def update_user(self, user_id: str, user_data: dict[str, Any]) -> None:
    """Update user details."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.reset_password abstractmethod

reset_password(
    user_id: str, password: str, temporary: bool = False
) -> None

Reset a user's password.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def reset_password(self, user_id: str, password: str, temporary: bool = False) -> None:
    """Reset a user's password."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.search_users abstractmethod

search_users(
    query: str, max_results: int = 100
) -> list[KeycloakUserType]

Search for users by username, email, or name.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def search_users(self, query: str, max_results: int = 100) -> list[KeycloakUserType]:
    """Search for users by username, email, or name."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.clear_user_sessions abstractmethod

clear_user_sessions(user_id: str) -> None

Clear all sessions for a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def clear_user_sessions(self, user_id: str) -> None:
    """Clear all sessions for a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_user_roles abstractmethod

get_user_roles(user_id: str) -> list[KeycloakRoleType]

Get roles assigned to a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_user_roles(self, user_id: str) -> list[KeycloakRoleType]:
    """Get roles assigned to a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_client_roles_for_user abstractmethod

get_client_roles_for_user(
    user_id: str, client_id: str
) -> list[KeycloakRoleType]

Get client-specific roles assigned to a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_client_roles_for_user(self, user_id: str, client_id: str) -> list[KeycloakRoleType]:
    """Get client-specific roles assigned to a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.has_role abstractmethod

has_role(token: str, role_name: str) -> bool

Check if a user has a specific role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def has_role(self, token: str, role_name: str) -> bool:
    """Check if a user has a specific role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.has_any_of_roles abstractmethod

has_any_of_roles(
    token: str, role_names: frozenset[str]
) -> bool

Check if a user has any of the specified roles.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def has_any_of_roles(self, token: str, role_names: frozenset[str]) -> bool:
    """Check if a user has any of the specified roles."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.has_all_roles abstractmethod

has_all_roles(
    token: str, role_names: frozenset[str]
) -> bool

Check if a user has all of the specified roles.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def has_all_roles(self, token: str, role_names: frozenset[str]) -> bool:
    """Check if a user has all of the specified roles."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.assign_realm_role abstractmethod

assign_realm_role(user_id: str, role_name: str) -> None

Assign a realm role to a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def assign_realm_role(self, user_id: str, role_name: str) -> None:
    """Assign a realm role to a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.remove_realm_role abstractmethod

remove_realm_role(user_id: str, role_name: str) -> None

Remove a realm role from a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def remove_realm_role(self, user_id: str, role_name: str) -> None:
    """Remove a realm role from a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.assign_client_role abstractmethod

assign_client_role(
    user_id: str, client_id: str, role_name: str
) -> None

Assign a client-specific role to a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def assign_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
    """Assign a client-specific role to a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.remove_client_role abstractmethod

remove_client_role(
    user_id: str, client_id: str, role_name: str
) -> None

Remove a client-specific role from a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def remove_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
    """Remove a client-specific role from a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_realm_role abstractmethod

get_realm_role(role_name: str) -> dict[str, Any]

Get realm role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_realm_role(self, role_name: str) -> dict[str, Any]:
    """Get realm role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_realm_roles abstractmethod

get_realm_roles() -> list[dict[str, Any]]

Get all realm roles.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_realm_roles(self) -> list[dict[str, Any]]:
    """Get all realm roles."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.create_realm_role abstractmethod

create_realm_role(
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None

Create a new realm role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def create_realm_role(
    self,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None:
    """Create a new realm role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.delete_realm_role abstractmethod

delete_realm_role(role_name: str) -> None

Delete a realm role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def delete_realm_role(self, role_name: str) -> None:
    """Delete a realm role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_client_id abstractmethod

get_client_id(client_name: str) -> str

Get client ID by client name.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_client_id(self, client_name: str) -> str:
    """Get client ID by client name."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_client_secret abstractmethod

get_client_secret(client_id: str) -> str

Get client secret.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_client_secret(self, client_id: str) -> str:
    """Get client secret."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_service_account_id abstractmethod

get_service_account_id() -> str

Get service account user ID for the current client.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_service_account_id(self) -> str:
    """Get service account user ID for the current client."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_public_key abstractmethod

get_public_key() -> PublicKeyType

Get the public key used to verify tokens.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_public_key(self) -> PublicKeyType:
    """Get the public key used to verify tokens."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_well_known_config abstractmethod

get_well_known_config() -> dict[str, Any]

Get the well-known OpenID configuration.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_well_known_config(self) -> dict[str, Any]:
    """Get the well-known OpenID configuration."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_certs abstractmethod

get_certs() -> dict[str, Any]

Get the JWT verification certificates.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_certs(self) -> dict[str, Any]:
    """Get the JWT verification certificates."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_token_from_code abstractmethod

get_token_from_code(
    code: str, redirect_uri: str
) -> KeycloakTokenType | None

Exchange authorization code for token.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_token_from_code(self, code: str, redirect_uri: str) -> KeycloakTokenType | None:
    """Exchange authorization code for token."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.check_permissions abstractmethod

check_permissions(
    token: str, resource: str, scope: str
) -> bool

Check if a user has permission to access a resource with the specified scope.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def check_permissions(self, token: str, resource: str, scope: str) -> bool:
    """Check if a user has permission to access a resource with the specified scope."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.delete_user abstractmethod

delete_user(user_id: str) -> None

Delete a user from Keycloak by their ID.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def delete_user(self, user_id: str) -> None:
    """Delete a user from Keycloak by their ID."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.create_client_role abstractmethod

create_client_role(
    client_id: str,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None

Create a new client role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def create_client_role(
    self,
    client_id: str,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None:
    """Create a new client role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.create_realm abstractmethod

create_realm(
    realm_name: str, skip_exists: bool = True, **kwargs: Any
) -> dict[str, Any] | None

Create a new Keycloak realm.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def create_realm(self, realm_name: str, skip_exists: bool = True, **kwargs: Any) -> dict[str, Any] | None:
    """Create a new Keycloak realm."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_realm abstractmethod

get_realm(realm_name: str) -> dict[str, Any] | None

Get realm details by realm name.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_realm(self, realm_name: str) -> dict[str, Any] | None:
    """Get realm details by realm name."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.update_realm abstractmethod

update_realm(
    realm_name: str, **kwargs: Any
) -> dict[str, Any] | None

Update a realm. Kwargs are RealmRepresentation top-level attributes (e.g. displayName, organizationsEnabled).

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def update_realm(self, realm_name: str, **kwargs: Any) -> dict[str, Any] | None:
    """Update a realm. Kwargs are RealmRepresentation top-level attributes (e.g. displayName, organizationsEnabled)."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.create_client abstractmethod

create_client(
    client_id: str,
    realm: str | None = None,
    skip_exists: bool = True,
    **kwargs: Any,
) -> dict[str, Any] | None

Create a new client in the specified realm.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def create_client(
    self,
    client_id: str,
    realm: str | None = None,
    skip_exists: bool = True,
    **kwargs: Any,
) -> dict[str, Any] | None:
    """Create a new client in the specified realm."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.add_realm_roles_to_composite abstractmethod

add_realm_roles_to_composite(
    composite_role_name: str, child_role_names: list[str]
) -> None

Add realm roles to a composite role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def add_realm_roles_to_composite(self, composite_role_name: str, child_role_names: list[str]) -> None:
    """Add realm roles to a composite role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.add_client_roles_to_composite abstractmethod

add_client_roles_to_composite(
    composite_role_name: str,
    client_id: str,
    child_role_names: list[str],
) -> None

Add client roles to a composite role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def add_client_roles_to_composite(
    self,
    composite_role_name: str,
    client_id: str,
    child_role_names: list[str],
) -> None:
    """Add client roles to a composite role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_composite_realm_roles abstractmethod

get_composite_realm_roles(
    role_name: str,
) -> list[dict[str, Any]] | None

Get composite roles for a realm role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_composite_realm_roles(self, role_name: str) -> list[dict[str, Any]] | None:
    """Get composite roles for a realm role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_organizations abstractmethod

get_organizations(
    query: dict | None = None,
) -> list[KeycloakOrganizationType]

Fetch all organizations. Returns list of OrganizationRepresentation, filtered by query.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_organizations(self, query: dict | None = None) -> list[KeycloakOrganizationType]:
    """Fetch all organizations. Returns list of OrganizationRepresentation, filtered by query."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_organization abstractmethod

get_organization(
    organization_id: str,
) -> KeycloakOrganizationType

Get representation of the organization by ID.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_organization(self, organization_id: str) -> KeycloakOrganizationType:
    """Get representation of the organization by ID."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.create_organization abstractmethod

create_organization(
    name: str, alias: str, **kwargs: Any
) -> str | None

Create a new organization. Name and alias must be unique. Returns org_id.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def create_organization(self, name: str, alias: str, **kwargs: Any) -> str | None:
    """Create a new organization. Name and alias must be unique. Returns org_id."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.update_organization abstractmethod

update_organization(
    organization_id: str, **kwargs: Any
) -> dict[str, Any]

Update an existing organization. Kwargs are organization attributes (e.g. name, alias).

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def update_organization(self, organization_id: str, **kwargs: Any) -> dict[str, Any]:
    """Update an existing organization. Kwargs are organization attributes (e.g. name, alias)."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.delete_organization abstractmethod

delete_organization(organization_id: str) -> dict[str, Any]

Delete an organization.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def delete_organization(self, organization_id: str) -> dict[str, Any]:
    """Delete an organization."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_organization_idps abstractmethod

get_organization_idps(
    organization_id: str,
) -> list[dict[str, Any]]

Get IDPs by organization id.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_organization_idps(self, organization_id: str) -> list[dict[str, Any]]:
    """Get IDPs by organization id."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_user_organizations abstractmethod

get_user_organizations(
    user_id: str,
) -> list[KeycloakOrganizationType]

Get organizations by user id. Returns list of organizations the user is member of.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_user_organizations(self, user_id: str) -> list[KeycloakOrganizationType]:
    """Get organizations by user id. Returns list of organizations the user is member of."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_organization_members abstractmethod

get_organization_members(
    organization_id: str, query: dict | None = None
) -> list[dict[str, Any]]

Get members by organization id, optionally filtered by query parameters.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_organization_members(self, organization_id: str, query: dict | None = None) -> list[dict[str, Any]]:
    """Get members by organization id, optionally filtered by query parameters."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.get_organization_members_count abstractmethod

get_organization_members_count(organization_id: str) -> int

Get the number of members in the organization.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_organization_members_count(self, organization_id: str) -> int:
    """Get the number of members in the organization."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.organization_user_add abstractmethod

organization_user_add(
    user_id: str, organization_id: str
) -> bytes

Add a user to an organization.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def organization_user_add(self, user_id: str, organization_id: str) -> bytes:
    """Add a user to an organization."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.KeycloakPort.organization_user_remove abstractmethod

organization_user_remove(
    user_id: str, organization_id: str
) -> dict[str, Any]

Remove a user from an organization.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def organization_user_remove(self, user_id: str, organization_id: str) -> dict[str, Any]:
    """Remove a user from an organization."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort

Asynchronous interface for Keycloak operations providing a standardized access pattern.

This interface defines the contract for async Keycloak adapters, ensuring consistent implementation of Keycloak operations across different adapters. It covers essential functionality including authentication, user management, and role management.

Source code in archipy/adapters/keycloak/ports.py

class AsyncKeycloakPort:
    """Asynchronous interface for Keycloak operations providing a standardized access pattern.

    This interface defines the contract for async Keycloak adapters, ensuring consistent
    implementation of Keycloak operations across different adapters. It covers essential
    functionality including authentication, user management, and role management.
    """

    # Token Operations
    @abstractmethod
    async def get_token(self, username: str, password: str) -> KeycloakTokenType | None:
        """Get a user token by username and password."""
        raise NotImplementedError

    @abstractmethod
    async def refresh_token(self, refresh_token: str) -> KeycloakTokenType | None:
        """Refresh an existing token using a refresh token."""
        raise NotImplementedError

    @abstractmethod
    async def validate_token(self, token: str) -> bool:
        """Validate if a token is still valid."""
        raise NotImplementedError

    @abstractmethod
    async def get_userinfo(self, token: str) -> KeycloakUserType | None:
        """Get user information from a token."""
        raise NotImplementedError

    @abstractmethod
    async def get_token_info(self, token: str) -> dict[str, Any] | None:
        """Decode token to get its claims."""
        raise NotImplementedError

    @abstractmethod
    async def introspect_token(self, token: str) -> dict[str, Any] | None:
        """Introspect token to get detailed information about it."""
        raise NotImplementedError

    @abstractmethod
    async def get_client_credentials_token(self) -> KeycloakTokenType | None:
        """Get token using client credentials."""
        raise NotImplementedError

    @abstractmethod
    async def logout(self, refresh_token: str) -> None:
        """Logout user by invalidating their refresh token."""
        raise NotImplementedError

    # User Operations
    @abstractmethod
    async def get_user_by_id(self, user_id: str) -> KeycloakUserType | None:
        """Get user details by user ID."""
        raise NotImplementedError

    @abstractmethod
    async def get_user_by_username(self, username: str) -> KeycloakUserType | None:
        """Get user details by username."""
        raise NotImplementedError

    @abstractmethod
    async def get_user_by_email(self, email: str) -> KeycloakUserType | None:
        """Get user details by email."""
        raise NotImplementedError

    @abstractmethod
    async def create_user(self, user_data: dict[str, Any]) -> str | None:
        """Create a new user in Keycloak."""
        raise NotImplementedError

    @abstractmethod
    async def update_user(self, user_id: str, user_data: dict[str, Any]) -> None:
        """Update user details."""
        raise NotImplementedError

    @abstractmethod
    async def reset_password(self, user_id: str, password: str, temporary: bool = False) -> None:
        """Reset a user's password."""
        raise NotImplementedError

    @abstractmethod
    async def search_users(self, query: str, max_results: int = 100) -> list[KeycloakUserType]:
        """Search for users by username, email, or name."""
        raise NotImplementedError

    @abstractmethod
    async def clear_user_sessions(self, user_id: str) -> None:
        """Clear all sessions for a user."""
        raise NotImplementedError

    # Role Operations
    @abstractmethod
    async def get_user_roles(self, user_id: str) -> list[KeycloakRoleType]:
        """Get roles assigned to a user."""
        raise NotImplementedError

    @abstractmethod
    async def get_client_roles_for_user(self, user_id: str, client_id: str) -> list[KeycloakRoleType]:
        """Get client-specific roles assigned to a user."""
        raise NotImplementedError

    @abstractmethod
    async def has_role(self, token: str, role_name: str) -> bool:
        """Check if a user has a specific role."""
        raise NotImplementedError

    @abstractmethod
    async def has_any_of_roles(self, token: str, role_names: frozenset[str]) -> bool:
        """Check if a user has any of the specified roles."""
        raise NotImplementedError

    @abstractmethod
    async def has_all_roles(self, token: str, role_names: frozenset[str]) -> bool:
        """Check if a user has all of the specified roles."""
        raise NotImplementedError

    @abstractmethod
    async def assign_realm_role(self, user_id: str, role_name: str) -> None:
        """Assign a realm role to a user."""
        raise NotImplementedError

    @abstractmethod
    async def remove_realm_role(self, user_id: str, role_name: str) -> None:
        """Remove a realm role from a user."""
        raise NotImplementedError

    @abstractmethod
    async def assign_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
        """Assign a client-specific role to a user."""
        raise NotImplementedError

    @abstractmethod
    async def remove_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
        """Remove a client-specific role from a user."""
        raise NotImplementedError

    @abstractmethod
    async def get_realm_role(self, role_name: str) -> dict[str, Any]:
        """Get realm role."""
        raise NotImplementedError

    @abstractmethod
    async def get_realm_roles(self) -> list[dict[str, Any]]:
        """Get all realm roles."""
        raise NotImplementedError

    @abstractmethod
    async def create_realm_role(
        self,
        role_name: str,
        description: str | None = None,
        skip_exists: bool = True,
    ) -> dict[str, Any] | None:
        """Create a new realm role."""
        raise NotImplementedError

    @abstractmethod
    async def delete_realm_role(self, role_name: str) -> None:
        """Delete a realm role."""
        raise NotImplementedError

    # Client Operations
    @abstractmethod
    async def get_client_id(self, client_name: str) -> str:
        """Get client ID by client name."""
        raise NotImplementedError

    @abstractmethod
    async def get_client_secret(self, client_id: str) -> str:
        """Get client secret."""
        raise NotImplementedError

    @abstractmethod
    async def get_service_account_id(self) -> str:
        """Get service account user ID for the current client."""
        raise NotImplementedError

    # System Operations
    @abstractmethod
    async def get_public_key(self) -> PublicKeyType:
        """Get the public key used to verify tokens."""
        raise NotImplementedError

    @abstractmethod
    async def get_well_known_config(self) -> dict[str, Any]:
        """Get the well-known OpenID configuration."""
        raise NotImplementedError

    @abstractmethod
    async def get_certs(self) -> dict[str, Any]:
        """Get the JWT verification certificates."""
        raise NotImplementedError

    # Authorization
    @abstractmethod
    async def get_token_from_code(self, code: str, redirect_uri: str) -> KeycloakTokenType | None:
        """Exchange authorization code for token."""
        raise NotImplementedError

    @abstractmethod
    async def check_permissions(self, token: str, resource: str, scope: str) -> bool:
        """Check if a user has permission to access a resource with the specified scope."""
        raise NotImplementedError

    @abstractmethod
    async def delete_user(self, user_id: str) -> None:
        """Delete a user from Keycloak by their ID."""
        raise NotImplementedError

    @abstractmethod
    async def create_client_role(
        self,
        client_id: str,
        role_name: str,
        description: str | None = None,
        skip_exists: bool = True,
    ) -> dict[str, Any] | None:
        """Create a new client role."""
        raise NotImplementedError

    @abstractmethod
    async def create_realm(self, realm_name: str, skip_exists: bool = True, **kwargs: Any) -> dict[str, Any] | None:
        """Create a new Keycloak realm."""
        raise NotImplementedError

    @abstractmethod
    async def get_realm(self, realm_name: str) -> dict[str, Any] | None:
        """Get realm details by realm name."""
        raise NotImplementedError

    @abstractmethod
    async def update_realm(self, realm_name: str, **kwargs: Any) -> dict[str, Any] | None:
        """Update a realm. Kwargs are RealmRepresentation top-level attributes (e.g. displayName, organizationsEnabled)."""
        raise NotImplementedError

    @abstractmethod
    async def create_client(
        self,
        client_id: str,
        realm: str | None = None,
        skip_exists: bool = True,
        **kwargs: Any,
    ) -> dict[str, Any] | None:
        """Create a new client in the specified realm."""
        raise NotImplementedError

    @abstractmethod
    async def add_realm_roles_to_composite(self, composite_role_name: str, child_role_names: list[str]) -> None:
        """Add realm roles to a composite role."""
        raise NotImplementedError

    @abstractmethod
    async def add_client_roles_to_composite(
        self,
        composite_role_name: str,
        client_id: str,
        child_role_names: list[str],
    ) -> None:
        """Add client roles to a composite role."""
        raise NotImplementedError

    @abstractmethod
    async def get_composite_realm_roles(self, role_name: str) -> list[dict[str, Any]] | None:
        """Get composite roles for a realm role."""
        raise NotImplementedError

    # Organization Operations
    @abstractmethod
    async def get_organizations(self, query: dict | None = None) -> list[KeycloakOrganizationType]:
        """Fetch all organizations. Returns list of OrganizationRepresentation, filtered by query."""
        raise NotImplementedError

    @abstractmethod
    async def get_organization(self, organization_id: str) -> KeycloakOrganizationType:
        """Get representation of the organization by ID."""
        raise NotImplementedError

    @abstractmethod
    async def create_organization(self, name: str, alias: str, **kwargs: Any) -> str | None:
        """Create a new organization. Name and alias must be unique. Returns org_id."""
        raise NotImplementedError

    @abstractmethod
    async def update_organization(self, organization_id: str, **kwargs: Any) -> dict[str, Any]:
        """Update an existing organization. Kwargs are organization attributes (e.g. name, alias)."""
        raise NotImplementedError

    @abstractmethod
    async def delete_organization(self, organization_id: str) -> dict[str, Any]:
        """Delete an organization."""
        raise NotImplementedError

    @abstractmethod
    async def get_organization_idps(self, organization_id: str) -> list[dict[str, Any]]:
        """Get IDPs by organization id."""
        raise NotImplementedError

    @abstractmethod
    async def get_user_organizations(self, user_id: str) -> list[KeycloakOrganizationType]:
        """Get organizations by user id. Returns list of organizations the user is member of."""
        raise NotImplementedError

    @abstractmethod
    async def get_organization_members(self, organization_id: str, query: dict | None = None) -> list[dict[str, Any]]:
        """Get members by organization id, optionally filtered by query parameters."""
        raise NotImplementedError

    @abstractmethod
    async def get_organization_members_count(self, organization_id: str) -> int:
        """Get the number of members in the organization."""
        raise NotImplementedError

    @abstractmethod
    async def organization_user_add(self, user_id: str, organization_id: str) -> bytes:
        """Add a user to an organization."""
        raise NotImplementedError

    @abstractmethod
    async def organization_user_remove(self, user_id: str, organization_id: str) -> dict[str, Any]:
        """Remove a user from an organization."""
        raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_token abstractmethod async

get_token(
    username: str, password: str
) -> KeycloakTokenType | None

Get a user token by username and password.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_token(self, username: str, password: str) -> KeycloakTokenType | None:
    """Get a user token by username and password."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.refresh_token abstractmethod async

refresh_token(
    refresh_token: str,
) -> KeycloakTokenType | None

Refresh an existing token using a refresh token.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def refresh_token(self, refresh_token: str) -> KeycloakTokenType | None:
    """Refresh an existing token using a refresh token."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.validate_token abstractmethod async

validate_token(token: str) -> bool

Validate if a token is still valid.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def validate_token(self, token: str) -> bool:
    """Validate if a token is still valid."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_userinfo abstractmethod async

get_userinfo(token: str) -> KeycloakUserType | None

Get user information from a token.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_userinfo(self, token: str) -> KeycloakUserType | None:
    """Get user information from a token."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_token_info abstractmethod async

get_token_info(token: str) -> dict[str, Any] | None

Decode token to get its claims.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_token_info(self, token: str) -> dict[str, Any] | None:
    """Decode token to get its claims."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.introspect_token abstractmethod async

introspect_token(token: str) -> dict[str, Any] | None

Introspect token to get detailed information about it.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def introspect_token(self, token: str) -> dict[str, Any] | None:
    """Introspect token to get detailed information about it."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_client_credentials_token abstractmethod async

get_client_credentials_token() -> KeycloakTokenType | None

Get token using client credentials.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_client_credentials_token(self) -> KeycloakTokenType | None:
    """Get token using client credentials."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.logout abstractmethod async

logout(refresh_token: str) -> None

Logout user by invalidating their refresh token.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def logout(self, refresh_token: str) -> None:
    """Logout user by invalidating their refresh token."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_user_by_id abstractmethod async

get_user_by_id(user_id: str) -> KeycloakUserType | None

Get user details by user ID.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_user_by_id(self, user_id: str) -> KeycloakUserType | None:
    """Get user details by user ID."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_user_by_username abstractmethod async

get_user_by_username(
    username: str,
) -> KeycloakUserType | None

Get user details by username.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_user_by_username(self, username: str) -> KeycloakUserType | None:
    """Get user details by username."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_user_by_email abstractmethod async

get_user_by_email(email: str) -> KeycloakUserType | None

Get user details by email.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_user_by_email(self, email: str) -> KeycloakUserType | None:
    """Get user details by email."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.create_user abstractmethod async

create_user(user_data: dict[str, Any]) -> str | None

Create a new user in Keycloak.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def create_user(self, user_data: dict[str, Any]) -> str | None:
    """Create a new user in Keycloak."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.update_user abstractmethod async

update_user(
    user_id: str, user_data: dict[str, Any]
) -> None

Update user details.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def update_user(self, user_id: str, user_data: dict[str, Any]) -> None:
    """Update user details."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.reset_password abstractmethod async

reset_password(
    user_id: str, password: str, temporary: bool = False
) -> None

Reset a user's password.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def reset_password(self, user_id: str, password: str, temporary: bool = False) -> None:
    """Reset a user's password."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.search_users abstractmethod async

search_users(
    query: str, max_results: int = 100
) -> list[KeycloakUserType]

Search for users by username, email, or name.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def search_users(self, query: str, max_results: int = 100) -> list[KeycloakUserType]:
    """Search for users by username, email, or name."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.clear_user_sessions abstractmethod async

clear_user_sessions(user_id: str) -> None

Clear all sessions for a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def clear_user_sessions(self, user_id: str) -> None:
    """Clear all sessions for a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_user_roles abstractmethod async

get_user_roles(user_id: str) -> list[KeycloakRoleType]

Get roles assigned to a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_user_roles(self, user_id: str) -> list[KeycloakRoleType]:
    """Get roles assigned to a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_client_roles_for_user abstractmethod async

get_client_roles_for_user(
    user_id: str, client_id: str
) -> list[KeycloakRoleType]

Get client-specific roles assigned to a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_client_roles_for_user(self, user_id: str, client_id: str) -> list[KeycloakRoleType]:
    """Get client-specific roles assigned to a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.has_role abstractmethod async

has_role(token: str, role_name: str) -> bool

Check if a user has a specific role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def has_role(self, token: str, role_name: str) -> bool:
    """Check if a user has a specific role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.has_any_of_roles abstractmethod async

has_any_of_roles(
    token: str, role_names: frozenset[str]
) -> bool

Check if a user has any of the specified roles.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def has_any_of_roles(self, token: str, role_names: frozenset[str]) -> bool:
    """Check if a user has any of the specified roles."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.has_all_roles abstractmethod async

has_all_roles(
    token: str, role_names: frozenset[str]
) -> bool

Check if a user has all of the specified roles.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def has_all_roles(self, token: str, role_names: frozenset[str]) -> bool:
    """Check if a user has all of the specified roles."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.assign_realm_role abstractmethod async

assign_realm_role(user_id: str, role_name: str) -> None

Assign a realm role to a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def assign_realm_role(self, user_id: str, role_name: str) -> None:
    """Assign a realm role to a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.remove_realm_role abstractmethod async

remove_realm_role(user_id: str, role_name: str) -> None

Remove a realm role from a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def remove_realm_role(self, user_id: str, role_name: str) -> None:
    """Remove a realm role from a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.assign_client_role abstractmethod async

assign_client_role(
    user_id: str, client_id: str, role_name: str
) -> None

Assign a client-specific role to a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def assign_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
    """Assign a client-specific role to a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.remove_client_role abstractmethod async

remove_client_role(
    user_id: str, client_id: str, role_name: str
) -> None

Remove a client-specific role from a user.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def remove_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
    """Remove a client-specific role from a user."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_realm_role abstractmethod async

get_realm_role(role_name: str) -> dict[str, Any]

Get realm role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_realm_role(self, role_name: str) -> dict[str, Any]:
    """Get realm role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_realm_roles abstractmethod async

get_realm_roles() -> list[dict[str, Any]]

Get all realm roles.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_realm_roles(self) -> list[dict[str, Any]]:
    """Get all realm roles."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.create_realm_role abstractmethod async

create_realm_role(
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None

Create a new realm role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def create_realm_role(
    self,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None:
    """Create a new realm role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.delete_realm_role abstractmethod async

delete_realm_role(role_name: str) -> None

Delete a realm role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def delete_realm_role(self, role_name: str) -> None:
    """Delete a realm role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_client_id abstractmethod async

get_client_id(client_name: str) -> str

Get client ID by client name.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_client_id(self, client_name: str) -> str:
    """Get client ID by client name."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_client_secret abstractmethod async

get_client_secret(client_id: str) -> str

Get client secret.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_client_secret(self, client_id: str) -> str:
    """Get client secret."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_service_account_id abstractmethod async

get_service_account_id() -> str

Get service account user ID for the current client.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_service_account_id(self) -> str:
    """Get service account user ID for the current client."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_public_key abstractmethod async

get_public_key() -> PublicKeyType

Get the public key used to verify tokens.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_public_key(self) -> PublicKeyType:
    """Get the public key used to verify tokens."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_well_known_config abstractmethod async

get_well_known_config() -> dict[str, Any]

Get the well-known OpenID configuration.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_well_known_config(self) -> dict[str, Any]:
    """Get the well-known OpenID configuration."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_certs abstractmethod async

get_certs() -> dict[str, Any]

Get the JWT verification certificates.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_certs(self) -> dict[str, Any]:
    """Get the JWT verification certificates."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_token_from_code abstractmethod async

get_token_from_code(
    code: str, redirect_uri: str
) -> KeycloakTokenType | None

Exchange authorization code for token.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_token_from_code(self, code: str, redirect_uri: str) -> KeycloakTokenType | None:
    """Exchange authorization code for token."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.check_permissions abstractmethod async

check_permissions(
    token: str, resource: str, scope: str
) -> bool

Check if a user has permission to access a resource with the specified scope.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def check_permissions(self, token: str, resource: str, scope: str) -> bool:
    """Check if a user has permission to access a resource with the specified scope."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.delete_user abstractmethod async

delete_user(user_id: str) -> None

Delete a user from Keycloak by their ID.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def delete_user(self, user_id: str) -> None:
    """Delete a user from Keycloak by their ID."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.create_client_role abstractmethod async

create_client_role(
    client_id: str,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None

Create a new client role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def create_client_role(
    self,
    client_id: str,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None:
    """Create a new client role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.create_realm abstractmethod async

create_realm(
    realm_name: str, skip_exists: bool = True, **kwargs: Any
) -> dict[str, Any] | None

Create a new Keycloak realm.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def create_realm(self, realm_name: str, skip_exists: bool = True, **kwargs: Any) -> dict[str, Any] | None:
    """Create a new Keycloak realm."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_realm abstractmethod async

get_realm(realm_name: str) -> dict[str, Any] | None

Get realm details by realm name.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_realm(self, realm_name: str) -> dict[str, Any] | None:
    """Get realm details by realm name."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.update_realm abstractmethod async

update_realm(
    realm_name: str, **kwargs: Any
) -> dict[str, Any] | None

Update a realm. Kwargs are RealmRepresentation top-level attributes (e.g. displayName, organizationsEnabled).

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def update_realm(self, realm_name: str, **kwargs: Any) -> dict[str, Any] | None:
    """Update a realm. Kwargs are RealmRepresentation top-level attributes (e.g. displayName, organizationsEnabled)."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.create_client abstractmethod async

create_client(
    client_id: str,
    realm: str | None = None,
    skip_exists: bool = True,
    **kwargs: Any,
) -> dict[str, Any] | None

Create a new client in the specified realm.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def create_client(
    self,
    client_id: str,
    realm: str | None = None,
    skip_exists: bool = True,
    **kwargs: Any,
) -> dict[str, Any] | None:
    """Create a new client in the specified realm."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.add_realm_roles_to_composite abstractmethod async

add_realm_roles_to_composite(
    composite_role_name: str, child_role_names: list[str]
) -> None

Add realm roles to a composite role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def add_realm_roles_to_composite(self, composite_role_name: str, child_role_names: list[str]) -> None:
    """Add realm roles to a composite role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.add_client_roles_to_composite abstractmethod async

add_client_roles_to_composite(
    composite_role_name: str,
    client_id: str,
    child_role_names: list[str],
) -> None

Add client roles to a composite role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def add_client_roles_to_composite(
    self,
    composite_role_name: str,
    client_id: str,
    child_role_names: list[str],
) -> None:
    """Add client roles to a composite role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_composite_realm_roles abstractmethod async

get_composite_realm_roles(
    role_name: str,
) -> list[dict[str, Any]] | None

Get composite roles for a realm role.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_composite_realm_roles(self, role_name: str) -> list[dict[str, Any]] | None:
    """Get composite roles for a realm role."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_organizations abstractmethod async

get_organizations(
    query: dict | None = None,
) -> list[KeycloakOrganizationType]

Fetch all organizations. Returns list of OrganizationRepresentation, filtered by query.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_organizations(self, query: dict | None = None) -> list[KeycloakOrganizationType]:
    """Fetch all organizations. Returns list of OrganizationRepresentation, filtered by query."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_organization abstractmethod async

get_organization(
    organization_id: str,
) -> KeycloakOrganizationType

Get representation of the organization by ID.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_organization(self, organization_id: str) -> KeycloakOrganizationType:
    """Get representation of the organization by ID."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.create_organization abstractmethod async

create_organization(
    name: str, alias: str, **kwargs: Any
) -> str | None

Create a new organization. Name and alias must be unique. Returns org_id.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def create_organization(self, name: str, alias: str, **kwargs: Any) -> str | None:
    """Create a new organization. Name and alias must be unique. Returns org_id."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.update_organization abstractmethod async

update_organization(
    organization_id: str, **kwargs: Any
) -> dict[str, Any]

Update an existing organization. Kwargs are organization attributes (e.g. name, alias).

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def update_organization(self, organization_id: str, **kwargs: Any) -> dict[str, Any]:
    """Update an existing organization. Kwargs are organization attributes (e.g. name, alias)."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.delete_organization abstractmethod async

delete_organization(organization_id: str) -> dict[str, Any]

Delete an organization.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def delete_organization(self, organization_id: str) -> dict[str, Any]:
    """Delete an organization."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_organization_idps abstractmethod async

get_organization_idps(
    organization_id: str,
) -> list[dict[str, Any]]

Get IDPs by organization id.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_organization_idps(self, organization_id: str) -> list[dict[str, Any]]:
    """Get IDPs by organization id."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_user_organizations abstractmethod async

get_user_organizations(
    user_id: str,
) -> list[KeycloakOrganizationType]

Get organizations by user id. Returns list of organizations the user is member of.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_user_organizations(self, user_id: str) -> list[KeycloakOrganizationType]:
    """Get organizations by user id. Returns list of organizations the user is member of."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_organization_members abstractmethod async

get_organization_members(
    organization_id: str, query: dict | None = None
) -> list[dict[str, Any]]

Get members by organization id, optionally filtered by query parameters.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_organization_members(self, organization_id: str, query: dict | None = None) -> list[dict[str, Any]]:
    """Get members by organization id, optionally filtered by query parameters."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.get_organization_members_count abstractmethod async

get_organization_members_count(organization_id: str) -> int

Get the number of members in the organization.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_organization_members_count(self, organization_id: str) -> int:
    """Get the number of members in the organization."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.organization_user_add abstractmethod async

organization_user_add(
    user_id: str, organization_id: str
) -> bytes

Add a user to an organization.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def organization_user_add(self, user_id: str, organization_id: str) -> bytes:
    """Add a user to an organization."""
    raise NotImplementedError

archipy.adapters.keycloak.ports.AsyncKeycloakPort.organization_user_remove abstractmethod async

organization_user_remove(
    user_id: str, organization_id: str
) -> dict[str, Any]

Remove a user from an organization.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def organization_user_remove(self, user_id: str, organization_id: str) -> dict[str, Any]:
    """Remove a user from an organization."""
    raise NotImplementedError

options: show_root_toc_entry: false heading_level: 3

Adapters

Concrete Keycloak adapter wrapping the Keycloak REST API for authentication and authorization operations.

archipy.adapters.keycloak.adapters.logger module-attribute

logger = getLogger(__name__)

archipy.adapters.keycloak.adapters.KeycloakExceptionHandlerMixin

Mixin class to handle Keycloak exceptions in a consistent way.

Source code in archipy/adapters/keycloak/adapters.py

class KeycloakExceptionHandlerMixin:
    """Mixin class to handle Keycloak exceptions in a consistent way."""

    @classmethod
    def _extract_error_message(cls, exception: KeycloakError) -> str:
        """Extract the actual error message from Keycloak error response.

        Args:
            exception: The Keycloak exception

        Returns:
            str: The extracted error message
        """
        error_message = str(exception)

        # Try to parse JSON response body
        if hasattr(exception, "response_body") and exception.response_body:
            try:
                body = exception.response_body
                if isinstance(body, bytes):
                    body_str = body.decode("utf-8")
                elif isinstance(body, str):
                    body_str = body
                else:
                    body_str = str(body)

                parsed = json.loads(body_str)
                if isinstance(parsed, dict):
                    error_message = (
                        parsed.get("errorMessage")
                        or parsed.get("error_description")
                        or parsed.get("error")
                        or error_message
                    )
            except json.JSONDecodeError, UnicodeDecodeError:
                pass

        return error_message

    @classmethod
    def _handle_keycloak_exception(cls, exception: KeycloakError, operation: str) -> NoReturn:
        """Handle Keycloak exceptions and map them to appropriate application errors.

        Args:
            exception: The original Keycloak exception
            operation: The name of the operation that failed

        Raises:
            Various application-specific errors based on the exception type/content
        """
        error_message = cls._extract_error_message(exception)
        response_code = getattr(exception, "response_code", None)
        error_lower = error_message.lower()

        # Common context data
        additional_data = {
            "operation": operation,
            "original_error": error_message,
            "response_code": response_code,
            "keycloak_error_type": type(exception).__name__,
        }

        # Connection and network errors
        if isinstance(exception, KeycloakConnectionError):
            if "timeout" in error_lower:
                raise KeycloakConnectionTimeoutError(additional_data=additional_data) from exception
            raise KeycloakServiceUnavailableError(additional_data=additional_data) from exception

        # Authentication errors
        if isinstance(exception, KeycloakAuthenticationError) or any(
            phrase in error_lower
            for phrase in ["invalid user credentials", "invalid credentials", "authentication failed", "unauthorized"]
        ):
            raise InvalidCredentialsError(additional_data=additional_data) from exception

        # Resource already exists errors
        if "already exists" in error_lower:
            if "realm" in error_lower:
                raise RealmAlreadyExistsError(additional_data=additional_data) from exception
            elif "user exists with same" in error_lower:
                raise UserAlreadyExistsError(additional_data=additional_data) from exception
            elif "client" in error_lower:
                raise ClientAlreadyExistsError(additional_data=additional_data) from exception
            elif "role" in error_lower:
                raise RoleAlreadyExistsError(additional_data=additional_data) from exception

        # Not found errors
        if "not found" in error_lower:
            raise ResourceNotFoundError(additional_data=additional_data) from exception

        # Permission errors
        if any(
            phrase in error_lower
            for phrase in ["forbidden", "access denied", "insufficient permissions", "insufficient scope"]
        ):
            raise InsufficientPermissionsError(additional_data=additional_data) from exception

        # Password policy errors
        if any(
            phrase in error_lower
            for phrase in ["invalid password", "password policy", "minimum length", "password must"]
        ):
            raise PasswordPolicyError(additional_data=additional_data) from exception

        # Validation errors (400 status codes that don't match above)
        if response_code == 400 or any(
            phrase in error_lower for phrase in ["validation", "invalid", "required field", "bad request"]
        ):
            raise ValidationError(additional_data=additional_data) from exception

        # Service unavailable
        if response_code in [503, 504] or "unavailable" in error_lower:
            raise KeycloakServiceUnavailableError(additional_data=additional_data) from exception

        # Default to InternalError for unrecognized errors
        raise InternalError(additional_data=additional_data) from exception

    @classmethod
    def _handle_realm_exception(
        cls,
        exception: KeycloakError,
        operation: str,
        realm_name: str | None = None,
    ) -> NoReturn:
        """Handle realm-specific exceptions.

        Args:
            exception: The original Keycloak exception
            operation: The name of the operation that failed
            realm_name: The realm name involved in the operation

        Raises:
            RealmAlreadyExistsError: If realm already exists
            Various other errors from _handle_keycloak_exception
        """
        # Add realm-specific context
        error_message = cls._extract_error_message(exception)

        # Realm-specific error handling
        if realm_name and "already exists" in error_message.lower():
            additional_data = {
                "operation": operation,
                "realm_name": realm_name,
                "original_error": error_message,
                "response_code": getattr(exception, "response_code", None),
            }
            raise RealmAlreadyExistsError(additional_data=additional_data) from exception

        # Fall back to general Keycloak error handling
        cls._handle_keycloak_exception(exception, operation)

    @classmethod
    def _handle_user_exception(
        cls,
        exception: KeycloakError,
        operation: str,
        user_data: dict | None = None,
    ) -> NoReturn:
        """Handle user-specific exceptions.

        Args:
            exception: The original Keycloak exception
            operation: The name of the operation that failed
            user_data: The user data involved in the operation

        Raises:
            UserAlreadyExistsError: If user already exists
            Various other errors from _handle_keycloak_exception
        """
        error_message = cls._extract_error_message(exception)

        # User-specific error handling
        if "user exists with same" in error_message.lower():
            additional_data = {
                "operation": operation,
                "original_error": error_message,
                "response_code": getattr(exception, "response_code", None),
            }
            if user_data:
                additional_data.update(
                    {
                        "username": user_data.get("username"),
                        "email": user_data.get("email"),
                    },
                )
            raise UserAlreadyExistsError(additional_data=additional_data) from exception

        # Fall back to general Keycloak error handling
        cls._handle_keycloak_exception(exception, operation)

    @classmethod
    def _handle_client_exception(
        cls,
        exception: KeycloakError,
        operation: str,
        client_data: dict | None = None,
    ) -> None:
        """Handle client-specific exceptions.

        Args:
            exception: The original Keycloak exception
            operation: The name of the operation that failed
            client_data: The client data involved in the operation

        Raises:
            ClientAlreadyExistsError: If client already exists
            Various other errors from _handle_keycloak_exception
        """
        error_message = cls._extract_error_message(exception)

        # Client-specific error handling
        if "client" in error_message.lower() and "already exists" in error_message.lower():
            additional_data = {
                "operation": operation,
                "original_error": error_message,
                "response_code": getattr(exception, "response_code", None),
            }
            if client_data:
                additional_data.update(
                    {
                        "client_id": client_data.get("clientId"),
                        "client_name": client_data.get("name"),
                    },
                )
            raise ClientAlreadyExistsError(additional_data=additional_data) from exception

        # Fall back to general Keycloak error handling
        cls._handle_keycloak_exception(exception, operation)

archipy.adapters.keycloak.adapters.KeycloakAdapter

Bases: KeycloakPort, KeycloakExceptionHandlerMixin

Concrete implementation of the KeycloakPort interface using python-keycloak library.

This implementation includes TTL caching for appropriate operations to improve performance while ensuring cache entries expire after a configured time to prevent stale data.

Source code in archipy/adapters/keycloak/adapters.py

class KeycloakAdapter(KeycloakPort, KeycloakExceptionHandlerMixin):
    """Concrete implementation of the KeycloakPort interface using python-keycloak library.

    This implementation includes TTL caching for appropriate operations to improve performance
    while ensuring cache entries expire after a configured time to prevent stale data.
    """

    def __init__(self, keycloak_configs: KeycloakConfig | None = None) -> None:
        """Initialize KeycloakAdapter with configuration.

        Args:
            keycloak_configs: Optional Keycloak configuration. If None, global config is used.
        """
        self.configs: KeycloakConfig = (
            BaseConfig.global_config().KEYCLOAK if keycloak_configs is None else keycloak_configs
        )

        # Initialize the OpenID client for authentication
        self._openid_adapter = self._get_openid_client(self.configs)

        # Cache for admin client to avoid unnecessary re-authentication
        self._admin_adapter: KeycloakAdmin | None = None
        self._admin_token_expiry: float = 0.0

        # Initialize admin client if admin mode is enabled and credentials are provided
        if self.configs.IS_ADMIN_MODE_ENABLED and (
            self.configs.CLIENT_SECRET_KEY or (self.configs.ADMIN_USERNAME and self.configs.ADMIN_PASSWORD)
        ):
            self._initialize_admin_client()

    def clear_all_caches(self) -> None:
        """Clear all cached values."""
        for attr_name in dir(self):
            attr = getattr(self, attr_name)
            if hasattr(attr, "clear_cache"):
                attr.clear_cache()

    @staticmethod
    def _get_openid_client(configs: KeycloakConfig) -> KeycloakOpenID:
        """Create and configure a KeycloakOpenID instance.

        Args:
            configs: Keycloak configuration

        Returns:
            Configured KeycloakOpenID client
        """
        server_url = configs.SERVER_URL
        client_id = configs.CLIENT_ID
        if not server_url or not client_id:
            raise ValueError("SERVER_URL and CLIENT_ID must be provided")
        return KeycloakOpenID(
            server_url=server_url,
            client_id=client_id,
            realm_name=configs.REALM_NAME,
            client_secret_key=configs.CLIENT_SECRET_KEY,
            verify=configs.VERIFY_SSL,
            timeout=configs.TIMEOUT,
        )

    def _initialize_admin_client(self) -> None:
        """Initialize or refresh the admin client."""
        try:
            # Check if admin credentials are available
            if self.configs.ADMIN_USERNAME and self.configs.ADMIN_PASSWORD:
                # Create admin client using admin credentials
                self._admin_adapter = KeycloakAdmin(
                    server_url=self.configs.SERVER_URL,
                    username=self.configs.ADMIN_USERNAME,
                    password=self.configs.ADMIN_PASSWORD,
                    realm_name=self.configs.REALM_NAME,
                    user_realm_name=self.configs.ADMIN_REALM_NAME,
                    verify=self.configs.VERIFY_SSL,
                    timeout=self.configs.TIMEOUT,
                )
                # Since we're using direct credentials, set a long expiry time
                self._admin_token_expiry = time.time() + 3600  # 1 hour
                logger.debug("Admin client initialized with admin credentials")

            elif self.configs.CLIENT_SECRET_KEY:
                # Get token using client credentials
                token = self._openid_adapter.token(grant_type="client_credentials")

                # Set token expiry time (current time + expires_in - buffer)
                # Using a 30-second buffer to ensure we refresh before expiration
                self._admin_token_expiry = time.time() + token.get("expires_in", 60) - 30

                self._admin_adapter = KeycloakAdmin(
                    server_url=self.configs.SERVER_URL,
                    realm_name=self.configs.REALM_NAME,
                    token=token,
                    verify=self.configs.VERIFY_SSL,
                    timeout=self.configs.TIMEOUT,
                )
                logger.debug("Admin client initialized with client credentials")

            else:
                raise UnauthenticatedError(
                    additional_data={"detail": "Neither admin credentials nor client secret provided"},
                )

        except KeycloakAuthenticationError as e:
            self._admin_adapter = None
            self._admin_token_expiry = 0
            raise UnauthenticatedError(
                additional_data={"detail": "Failed to authenticate with Keycloak service account"},
            ) from e
        except KeycloakConnectionError as e:
            self._admin_adapter = None
            self._admin_token_expiry = 0
            raise ConnectionTimeoutError("Failed to connect to Keycloak server") from e
        except KeycloakError as e:
            self._admin_adapter = None
            self._admin_token_expiry = 0
            self._handle_keycloak_exception(e, "_initialize_admin_client")

    @property
    def admin_adapter(self) -> KeycloakAdmin:
        """Get the admin adapter, refreshing it if necessary.

        Returns:
            KeycloakAdmin instance

        Raises:
            UnauthenticatedError: If admin client is not available due to authentication issues
            UnavailableError: If Keycloak service is unavailable
        """
        if not self.configs.IS_ADMIN_MODE_ENABLED or not (
            self.configs.CLIENT_SECRET_KEY or (self.configs.ADMIN_USERNAME and self.configs.ADMIN_PASSWORD)
        ):
            raise UnauthenticatedError(
                additional_data={
                    "data": "Admin mode is disabled or neither admin credentials nor client secret provided",
                },
            )

        # Check if token is about to expire and refresh if needed
        if self._admin_adapter is None or time.time() >= self._admin_token_expiry:
            self._initialize_admin_client()

        if self._admin_adapter is None:
            raise UnavailableError("Keycloak admin client is not available")

        return self._admin_adapter

    @override
    @ttl_cache_decorator(ttl_seconds=3600, maxsize=1)  # Cache for 1 hour, public key rarely changes
    def get_public_key(self) -> PublicKeyType:
        """Get the public key used to verify tokens.

        Returns:
            JWK key object used to verify signatures

        Raises:
            ServiceUnavailableError: If Keycloak service is unavailable
            InternalError: If there's an internal error processing the public key
        """
        try:
            from jwcrypto import jwk

            keys_info = self._openid_adapter.public_key()
            key = f"-----BEGIN PUBLIC KEY-----\n{keys_info}\n-----END PUBLIC KEY-----"
            return jwk.JWK.from_pem(key.encode("utf-8"))
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_public_key")
        except Exception as e:
            raise InternalError(additional_data={"operation": "get_public_key", "error": str(e)}) from e

    @override
    def get_token(self, username: str, password: str) -> KeycloakTokenType | None:
        """Get a user token by username and password using the Resource Owner Password Credentials Grant.

        Warning:
            This method uses the direct password grant flow, which is less secure and not recommended
            for user login in production environments. Instead, prefer the web-based OAuth 2.0
            Authorization Code Flow (use `get_token_from_code`) for secure authentication.
            Use this method only for testing, administrative tasks, or specific service accounts
            where direct credential use is acceptable and properly secured.

        Args:
            username: User's username
            password: User's password

        Returns:
            Token response containing access_token, refresh_token, etc.

        Raises:
            InvalidCredentialsError: If username or password is invalid
            ServiceUnavailableError: If Keycloak service is unavailable
        """
        try:
            return self._openid_adapter.token(grant_type="password", username=username, password=password)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_token")

    @override
    def refresh_token(self, refresh_token: str) -> KeycloakTokenType | None:
        """Refresh an existing token using a refresh token.

        Args:
            refresh_token: Refresh token string

        Returns:
            New token response containing access_token, refresh_token, etc.

        Raises:
            InvalidTokenError: If refresh token is invalid or expired
            ServiceUnavailableError: If Keycloak service is unavailable
        """
        try:
            return self._openid_adapter.refresh_token(refresh_token)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "refresh_token")

    @override
    def validate_token(self, token: str) -> bool:
        """Validate if a token is still valid.

        Args:
            token: Access token to validate

        Returns:
            True if token is valid, False otherwise
        """
        # Not caching validation results as tokens are time-sensitive
        try:
            # Let the underlying adapter handle key selection to align with expected types
            self._openid_adapter.decode_token(token)
        except Exception as e:
            logger.debug(f"Token validation failed: {e!s}")
            return False
        else:
            return True

    @override
    def get_userinfo(self, token: str) -> KeycloakUserType | None:
        """Get user information from a token.

        Args:
            token: Access token

        Returns:
            User information

        Raises:
            ValueError: If getting user info fails
        """
        if not self.validate_token(token):
            raise InvalidTokenError()
        try:
            # _get_userinfo_cached returns KeycloakUserType (dict[str, Any])
            # The ttl_cache_decorator loses type info, but runtime behavior is correct
            # Access underlying function for proper typing
            cached_func = self._get_userinfo_cached
            underlying_func = getattr(cached_func, "__wrapped__", None)
            if underlying_func is not None:
                # Call underlying function directly for type checking
                result: KeycloakUserType = underlying_func(self, token)
            else:
                # Fallback to cached version if __wrapped__ not available
                result_raw = cached_func(token)
                if not isinstance(result_raw, dict):
                    return None
                # Type assertion: result_raw is a dict, which matches KeycloakUserType
                # Convert to proper type by creating a new dict with explicit typing
                result: KeycloakUserType = {str(k): v for k, v in result_raw.items()}
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_userinfo")
            return None
        else:
            return result

    @ttl_cache_decorator(ttl_seconds=30, maxsize=100)  # Cache for 30 seconds
    def _get_userinfo_cached(self, token: str) -> KeycloakUserType:
        return self._openid_adapter.userinfo(token)  # type: ignore[return-value]

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)  # Cache for 5 minutes
    def get_user_by_id(self, user_id: str) -> KeycloakUserType | None:
        """Get user details by user ID.

        Args:
            user_id: User's ID

        Returns:
            User details or None if not found

        Raises:
            ValueError: If getting user fails
        """
        try:
            return self.admin_adapter.get_user(user_id)
        except KeycloakGetError as e:
            if e.response_code == 404:
                return None
            self._handle_keycloak_exception(e, "get_user_by_id")
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_user_by_id")

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)  # Cache for 5 minutes
    def get_user_by_username(self, username: str) -> KeycloakUserType | None:
        """Get user details by username.

        Args:
            username: User's username

        Returns:
            User details or None if not found

        Raises:
            ValueError: If query fails
        """
        try:
            users = self.admin_adapter.get_users({"username": username})
            return users[0] if users else None
        except KeycloakError as e:
            raise InternalError() from e

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)  # Cache for 5 minutes
    def get_user_by_email(self, email: str) -> KeycloakUserType | None:
        """Get user details by email.

        Args:
            email: User's email

        Returns:
            User details or None if not found

        Raises:
            ValueError: If query fails
        """
        try:
            users = self.admin_adapter.get_users({"email": email})
            return users[0] if users else None
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_user_by_email")

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)  # Cache for 5 minutes
    def get_user_roles(self, user_id: str) -> list[KeycloakRoleType] | None:
        """Get roles assigned to a user.

        Args:
            user_id: User's ID

        Returns:
            List of roles

        Raises:
            ValueError: If getting roles fails
        """
        try:
            return self.admin_adapter.get_realm_roles_of_user(user_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_user_roles")

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)  # Cache for 5 minutes
    def get_client_roles_for_user(self, user_id: str, client_id: str) -> list[KeycloakRoleType]:
        """Get client-specific roles assigned to a user.

        Args:
            user_id: User's ID
            client_id: Client ID

        Returns:
            List of client-specific roles

        Raises:
            ValueError: If getting roles fails
        """
        try:
            return self.admin_adapter.get_client_roles_of_user(user_id, client_id)
        except KeycloakError as e:
            raise InternalError() from e

    @override
    def create_user(self, user_data: dict[str, Any]) -> str | None:
        """Create a new user in Keycloak.

        Args:
            user_data: User data including username, email, etc.

        Returns:
            ID of the created user

        Raises:
            ValueError: If creating user fails
        """
        # This is a write operation, no caching needed
        try:
            user_id = self.admin_adapter.create_user(user_data)

            # Clear related caches
            self.clear_all_caches()

        except KeycloakError as e:
            self._handle_user_exception(e, "create_user", user_data)
        else:
            return user_id

    @override
    def update_user(self, user_id: str, user_data: dict[str, Any]) -> None:
        """Update user details.

        Args:
            user_id: User's ID
            user_data: User data to update

        Raises:
            ValueError: If updating user fails
        """
        # This is a write operation, no caching needed
        try:
            self.admin_adapter.update_user(user_id, user_data)

            # Clear user-related caches
            self.clear_all_caches()

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "update_user")

    @override
    def reset_password(self, user_id: str, password: str, temporary: bool = False) -> None:
        """Reset a user's password.

        Args:
            user_id: User's ID
            password: New password
            temporary: Whether the password is temporary and should be changed on next login

        Raises:
            ValueError: If password reset fails
        """
        # This is a write operation, no caching needed
        try:
            self.admin_adapter.set_user_password(user_id, password, temporary)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "reset_password")

    @override
    def assign_realm_role(self, user_id: str, role_name: str) -> None:
        """Assign a realm role to a user.

        Args:
            user_id: User's ID
            role_name: Role name to assign

        Raises:
            ValueError: If role assignment fails
        """
        # This is a write operation, no caching needed
        try:
            # Get role representation
            role = self.admin_adapter.get_realm_role(role_name)
            # Assign role to user
            self.admin_adapter.assign_realm_roles(user_id, [role])

            # Clear role-related caches
            if hasattr(self.get_user_roles, "clear_cache"):
                self.get_user_roles.clear_cache()

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "assign_realm_role")

    @override
    def remove_realm_role(self, user_id: str, role_name: str) -> None:
        """Remove a realm role from a user.

        Args:
            user_id: User's ID
            role_name: Role name to remove

        Raises:
            ValueError: If role removal fails
        """
        # This is a write operation, no caching needed
        try:
            # Get role representation
            role = self.admin_adapter.get_realm_role(role_name)
            # Remove role from user
            self.admin_adapter.delete_realm_roles_of_user(user_id, [role])

            # Clear role-related caches
            if hasattr(self.get_user_roles, "clear_cache"):
                self.get_user_roles.clear_cache()

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "remove_realm_role")

    @override
    def assign_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
        """Assign a client-specific role to a user.

        Args:
            user_id: User's ID
            client_id: Client ID
            role_name: Role name to assign

        Raises:
            ValueError: If role assignment fails
        """
        # This is a write operation, no caching needed
        try:
            # Get client
            client = self.admin_adapter.get_client_id(client_id)
            if client is None:
                raise ValueError("client_id resolved to None")
            # Get role representation
            # Keycloak admin adapter methods accept these types at runtime
            role = self.admin_adapter.get_client_role(client, role_name)
            # Assign role to user
            self.admin_adapter.assign_client_role(user_id, client, [role])

            # Clear role-related caches
            if hasattr(self.get_client_roles_for_user, "clear_cache"):
                self.get_client_roles_for_user.clear_cache()

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "assign_client_role")

    @override
    def create_realm_role(
        self,
        role_name: str,
        description: str | None = None,
        skip_exists: bool = True,
    ) -> dict[str, Any] | None:
        """Create a new realm role.

        Args:
            role_name: Role name
            description: Optional role description
            skip_exists: Skip creation if realm role already exists

        Returns:
            Created role details

        Raises:
            ValueError: If role creation fails
        """
        # This is a write operation, no caching needed
        try:
            role_data = {"name": role_name}
            if description:
                role_data["description"] = description

            self.admin_adapter.create_realm_role(role_data, skip_exists=skip_exists)

            # Clear realm roles cache
            if hasattr(self.get_realm_roles, "clear_cache"):
                self.get_realm_roles.clear_cache()

            return self.admin_adapter.get_realm_role(role_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "create_realm_role")

    @override
    def create_client_role(
        self,
        client_id: str,
        role_name: str,
        description: str | None = None,
        skip_exists: bool = True,
    ) -> dict[str, Any] | None:
        """Create a new client role.

        Args:
            client_id: Client ID or client name
            role_name: Role name
            description: Optional role description
            skip_exists: Skip creation if client role already exists

        Returns:
            Created role details

        Raises:
            ValueError: If role creation fails
        """
        # This is a write operation, no caching needed
        try:
            resolved_client_id = self.admin_adapter.get_client_id(client_id)
            if resolved_client_id is None:
                raise ValueError(f"Client ID not found: {client_id}")

            # Prepare role data
            role_data = {"name": role_name}
            if description:
                role_data["description"] = description

            # Create client role
            self.admin_adapter.create_client_role(resolved_client_id, role_data, skip_exists=skip_exists)

            # Clear related caches if they exist
            if hasattr(self.get_client_roles_for_user, "clear_cache"):
                self.get_client_roles_for_user.clear_cache()

            # Return created role
            return self.admin_adapter.get_client_role(resolved_client_id, role_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "create_client_role")

    @override
    def delete_realm_role(self, role_name: str) -> None:
        """Delete a realm role.

        Args:
            role_name: Role name to delete

        Raises:
            ValueError: If role deletion fails
        """
        # This is a write operation, no caching needed
        try:
            self.admin_adapter.delete_realm_role(role_name)

            # Clear realm roles cache
            if hasattr(self.get_realm_roles, "clear_cache"):
                self.get_realm_roles.clear_cache()

            # We also need to clear user role caches since they might contain this role
            if hasattr(self.get_user_roles, "clear_cache"):
                self.get_user_roles.clear_cache()

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "delete_realm_role")

    @override
    @ttl_cache_decorator(ttl_seconds=3600, maxsize=1)  # Cache for 1 hour
    def get_service_account_id(self) -> str | None:
        """Get service account user ID for the current client.

        Returns:
            Service account user ID

        Raises:
            ValueError: If getting service account fails
        """
        try:
            client_id = self.get_client_id(self.configs.CLIENT_ID)
            return self.admin_adapter.get_client_service_account_user(str(client_id)).get("id")
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_service_account_id")

    @override
    @ttl_cache_decorator(ttl_seconds=3600, maxsize=1)  # Cache for 1 hour
    def get_well_known_config(self) -> dict[str, Any] | None:
        """Get the well-known OpenID configuration.

        Returns:
            OIDC configuration

        Raises:
            ValueError: If getting configuration fails
        """
        try:
            return self._openid_adapter.well_known()
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_well_known_config")

    @override
    @ttl_cache_decorator(ttl_seconds=3600, maxsize=1)  # Cache for 1 hour
    def get_certs(self) -> dict[str, Any] | None:
        """Get the JWT verification certificates.

        Returns:
            Certificate information

        Raises:
            ValueError: If getting certificates fails
        """
        try:
            return self._openid_adapter.certs()
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_certs")

    @override
    def get_token_from_code(self, code: str, redirect_uri: str) -> KeycloakTokenType | None:
        """Exchange authorization code for token.

        Args:
            code: Authorization code
            redirect_uri: Redirect URI used in authorization request

        Returns:
            Token response

        Raises:
            ValueError: If token exchange fails
        """
        # Authorization codes can only be used once, don't cache
        try:
            return self._openid_adapter.token(grant_type="authorization_code", code=code, redirect_uri=redirect_uri)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_token_from_code")

    @override
    def get_client_credentials_token(self) -> KeycloakTokenType | None:
        """Get token using client credentials.

        Returns:
            Token response

        Raises:
            ValueError: If token acquisition fails
        """
        # Tokens are time-sensitive, don't cache
        try:
            return self._openid_adapter.token(grant_type="client_credentials")
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_client_credentials_token")

    @override
    @ttl_cache_decorator(ttl_seconds=30, maxsize=50)  # Cache for 30 seconds with limited entries
    def search_users(self, query: str, max_results: int = 100) -> list[KeycloakUserType] | None:
        """Search for users by username, email, or name.

        Args:
            query: Search query
            max_results: Maximum number of results to return

        Returns:
            List of matching users

        Raises:
            ValueError: If search fails
        """
        try:
            # Try searching by different fields
            users = []

            # Search by username
            users.extend(self.admin_adapter.get_users({"username": query, "max": max_results}))

            # Search by email if no results or incomplete results
            if len(users) < max_results:
                remaining = max_results - len(users)
                email_users = self.admin_adapter.get_users({"email": query, "max": remaining})
                # Filter out duplicates
                user_ids = {user["id"] for user in users}
                users.extend([user for user in email_users if user["id"] not in user_ids])

            # Search by firstName if no results or incomplete results
            if len(users) < max_results:
                remaining = max_results - len(users)
                first_name_users = self.admin_adapter.get_users({"firstName": query, "max": remaining})
                # Filter out duplicates
                user_ids = {user["id"] for user in users}
                users.extend([user for user in first_name_users if user["id"] not in user_ids])

            # Search by lastName if no results or incomplete results
            if len(users) < max_results:
                remaining = max_results - len(users)
                last_name_users = self.admin_adapter.get_users({"lastName": query, "max": remaining})
                # Filter out duplicates
                user_ids = {user["id"] for user in users}
                users.extend([user for user in last_name_users if user["id"] not in user_ids])

            return users[:max_results]
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "search_users")

    @override
    @ttl_cache_decorator(ttl_seconds=3600, maxsize=50)  # Cache for 1 hour
    def get_client_secret(self, client_id: str) -> str | None:
        """Get client secret.

        Args:
            client_id: Client ID

        Returns:
            Client secret

        Raises:
            ValueError: If getting secret fails
        """
        try:
            client = self.admin_adapter.get_client(client_id)
            return client.get("secret", "")
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_client_secret")

    @override
    @ttl_cache_decorator(ttl_seconds=3600, maxsize=50)  # Cache for 1 hour
    def get_client_id(self, client_name: str) -> str | None:
        """Get client ID by client name.

        Args:
            client_name: Name of the client

        Returns:
            Client ID

        Raises:
            ValueError: If client not found
        """
        try:
            return self.admin_adapter.get_client_id(client_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_client_id")

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=1)  # Cache for 5 minutes
    def get_realm_roles(self) -> list[dict[str, Any]] | None:
        """Get all realm roles.

        Returns:
            List of realm roles

        Raises:
            ValueError: If getting roles fails
        """
        try:
            return self.admin_adapter.get_realm_roles()
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_realm_roles")

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=1)  # Cache for 5 minutes
    def get_realm_role(self, role_name: str) -> dict | None:
        """Get realm role.

        Args:
            role_name: Role name
        Returns:
            A realm role

        Raises:
            ValueError: If getting role fails
        """
        try:
            return self.admin_adapter.get_realm_role(role_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_realm_role")

    @override
    def remove_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
        """Remove a client-specific role from a user.

        Args:
            user_id: User's ID
            client_id: Client ID
            role_name: Role name to remove

        Raises:
            ValueError: If role removal fails
        """
        try:
            client = self.admin_adapter.get_client_id(client_id)
            if client is None:
                raise ValueError("client_id resolved to None")
            # Keycloak admin adapter methods accept these types at runtime
            role = self.admin_adapter.get_client_role(client, role_name)
            self.admin_adapter.delete_client_roles_of_user(user_id, client, [role])

            if hasattr(self.get_client_roles_for_user, "clear_cache"):
                self.get_client_roles_for_user.clear_cache()
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "remove_client_role")

    @override
    def clear_user_sessions(self, user_id: str) -> None:
        """Clear all sessions for a user.

        Args:
            user_id: User's ID

        Raises:
            ValueError: If clearing sessions fails
        """
        try:
            self.admin_adapter.user_logout(user_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "clear_user_sessions")

    @override
    def logout(self, refresh_token: str) -> None:
        """Logout user by invalidating their refresh token.

        Args:
            refresh_token: Refresh token to invalidate

        Raises:
            ValueError: If logout fails
        """
        try:
            self._openid_adapter.logout(refresh_token)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "logout")

    @override
    def introspect_token(self, token: str) -> dict[str, Any] | None:
        """Introspect token to get detailed information about it.

        Args:
            token: Access token

        Returns:
            Token introspection details

        Raises:
            ValueError: If token introspection fails
        """
        try:
            return self._openid_adapter.introspect(token)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "introspect_token")

    @override
    def get_token_info(self, token: str) -> dict[str, Any] | None:
        """Decode token to get its claims.

        Args:
            token: Access token

        Returns:
            Dictionary of token claims

        Raises:
            ValueError: If token decoding fails
        """
        try:
            # Let the underlying adapter handle key selection to align with expected types
            return self._openid_adapter.decode_token(token)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_token_info")

    @override
    def delete_user(self, user_id: str) -> None:
        """Delete a user from Keycloak by their ID.

        Args:
            user_id: The ID of the user to delete

        Raises:
            ValueError: If the deletion fails
        """
        try:
            self.admin_adapter.delete_user(user_id=user_id)

            if hasattr(self.get_user_by_username, "clear_cache"):
                self.get_user_by_username.clear_cache()

            logger.info(f"Successfully deleted user with ID {user_id}")
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "delete_user")

    @override
    def has_role(self, token: str, role_name: str) -> bool:
        """Check if a user has a specific role.

        Args:
            token: Access token
            role_name: Role name to check

        Returns:
            True if user has the role, False otherwise
        """
        # Not caching this result as token validation is time-sensitive
        try:
            user_info = self.get_userinfo(token)
            if not user_info:
                return False

            # Check realm roles
            realm_access = user_info.get("realm_access", {})
            roles = realm_access.get("roles", [])
            if role_name in roles:
                return True

            # Check client roles
            resource_access = user_info.get("resource_access", {})
            client_roles = resource_access.get(self.configs.CLIENT_ID, {}).get("roles", [])
            if role_name in client_roles:
                return True

        except Exception as e:
            logger.debug(f"Role check failed: {e!s}")
            return False
        else:
            return False

    @override
    def has_any_of_roles(self, token: str, role_names: frozenset[str]) -> bool:
        """Check if a user has any of the specified roles.

        Args:
            token: Access token
            role_names: Set of role names to check

        Returns:
            True if user has any of the roles, False otherwise
        """
        try:
            user_info = self.get_userinfo(token)
            if not user_info:
                return False

            # Check realm roles first
            realm_access = user_info.get("realm_access", {})
            realm_roles = set(realm_access.get("roles", []))
            if role_names.intersection(realm_roles):
                return True

            # Check roles for the configured client
            resource_access = user_info.get("resource_access", {})
            client_roles = set(resource_access.get(self.configs.CLIENT_ID, {}).get("roles", []))
            if role_names.intersection(client_roles):
                return True

        except Exception as e:
            logger.debug(f"Role check failed: {e!s}")
            return False
        else:
            return False

    @override
    def has_all_roles(self, token: str, role_names: frozenset[str]) -> bool:
        """Check if a user has all the specified roles.

        Args:
            token: Access token
            role_names: Set of role names to check

        Returns:
            True if user has all the roles, False otherwise
        """
        try:
            user_info = self.get_userinfo(token)
            if not user_info:
                return False

            # Get all user roles
            all_roles = set()

            # Add realm roles
            realm_access = user_info.get("realm_access", {})
            all_roles.update(realm_access.get("roles", []))

            # Add client roles
            resource_access = user_info.get("resource_access", {})
            client_roles = resource_access.get(self.configs.CLIENT_ID, {}).get("roles", [])
            all_roles.update(client_roles)

            # Check if all required roles are present
            return role_names.issubset(all_roles)

        except Exception as e:
            logger.debug(f"All roles check failed: {e!s}")
            return False

    @override
    def check_permissions(self, token: str, resource: str, scope: str) -> bool:
        """Check if a user has permission to access a resource with the specified scope.

        Args:
            token: Access token
            resource: Resource name
            scope: Permission scope

        Returns:
            True if permission granted, False otherwise
        """
        try:
            # Use UMA permissions endpoint to check specific resource and scope
            permissions = self._openid_adapter.uma_permissions(token, permissions=f"{resource}#{scope}")

            # Check if the response indicates permission is granted
            if not permissions or not isinstance(permissions, list):
                logger.debug("No permissions returned or invalid response format")
                return False

            # Look for the specific permission in the response
            for perm in permissions:
                if perm.get("rsname") == resource and scope in perm.get("scopes", []):
                    return True

        except KeycloakError as e:
            logger.debug(f"Permission check failed with Keycloak error: {e!s}")
            return False
        except Exception as e:
            logger.debug(f"Permission check failed with unexpected error: {e!s}")
            return False
        else:
            return False

    @override
    def create_realm(self, realm_name: str, skip_exists: bool = True, **kwargs: Any) -> dict[str, Any] | None:
        """Create a Keycloak realm with minimum required fields and optional additional config.

        Args:
            realm_name: The realm identifier (required)
            skip_exists: Skip creation if realm already exists
            kwargs: Additional optional configurations for the realm

        Returns:
            Realm details
        """
        payload = {
            "realm": realm_name,
            "enabled": kwargs.get("enabled", True),
            "displayName": kwargs.get("display_name", realm_name),
        }

        # Add any additional parameters from kwargs
        for key, value in kwargs.items():
            # Skip display_name as it's already handled
            if key == "display_name":
                continue

            # Convert Python snake_case to Keycloak camelCase
            camel_key = StringUtils.snake_to_camel_case(key)
            payload[camel_key] = value

        try:
            self.admin_adapter.create_realm(payload=payload, skip_exists=skip_exists)
        except KeycloakError as e:
            logger.debug(f"Failed to create realm: {e!s}")

            # Handle realm already exists with skip_exists option
            if skip_exists:
                error_message = self._extract_error_message(e).lower()
                if "already exists" in error_message and "realm" in error_message:
                    return {"realm": realm_name, "status": "already_exists", "config": payload}

            # Use the mixin to handle realm-specific errors
            self._handle_realm_exception(e, "create_realm", realm_name)
        else:
            return {"realm": realm_name, "status": "created", "config": payload}

    @override
    def get_realm(self, realm_name: str) -> dict[str, Any] | None:
        """Get realm details by realm name.

        Args:
            realm_name: Name of the realm

        Returns:
            Realm details
        """
        try:
            return self.admin_adapter.get_realm(realm_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_realm")

    @override
    def update_realm(self, realm_name: str, **kwargs: Any) -> dict[str, Any] | None:
        """Update a realm. Kwargs are RealmRepresentation.

        Args:
            realm_name: Realm name (not the realm id).
            **kwargs: RealmRepresentation attributes to update (e.g. displayName).

        Returns:
            Response from Keycloak, or None on error (handled via exception).
        """
        try:
            return self.admin_adapter.update_realm(realm_name, dict(kwargs))
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "update_realm")

    @override
    def create_client(
        self,
        client_id: str,
        realm: str | None = None,
        skip_exists: bool = True,
        **kwargs: Any,
    ) -> dict[str, Any] | None:
        """Create a Keycloak client with minimum required fields and optional additional config.

        Args:
            client_id: The client identifier (required)
            realm: Target realm name (uses the current realm in KeycloakAdmin if not specified)
            skip_exists: Skip creation if client already exists
            kwargs: Additional optional configurations for the client

        Returns:
            Client details
        """
        original_realm = self.admin_adapter.connection.realm_name

        try:
            # Set the target realm if provided
            if realm and realm != original_realm:
                self.admin_adapter.connection.realm_name = realm

            public_client = kwargs.get("public_client", False)

            # Prepare the minimal client payload
            payload = {
                "clientId": client_id,
                "enabled": kwargs.get("enabled", True),
                "protocol": kwargs.get("protocol", "openid-connect"),
                "name": kwargs.get("name", client_id),
                "publicClient": public_client,
            }

            # Enable service accounts for confidential clients by default
            if not public_client:
                payload["serviceAccountsEnabled"] = kwargs.get("service_account_enabled", True)
                payload["clientAuthenticatorType"] = "client-secret"

            for key, value in kwargs.items():
                if key in ["enabled", "protocol", "name", "public_client", "service_account_enabled"]:
                    continue

                # Convert snake_case to camelCase
                camel_key = StringUtils.snake_to_camel_case(key)
                payload[camel_key] = value

            internal_client_id = None
            try:
                internal_client_id = self.admin_adapter.create_client(payload, skip_exists=skip_exists)
            except KeycloakError as e:
                logger.debug(f"Failed to create client: {e!s}")

                # Handle client already exists with skip_exists option
                if skip_exists:
                    error_message = self._extract_error_message(e).lower()
                    if "already exists" in error_message and "client" in error_message:
                        return {
                            "client_id": client_id,
                            "status": "already_exists",
                            "realm": self.admin_adapter.connection.realm_name,
                        }

                # Use the mixin to handle client-specific errors
                client_data = {"clientId": client_id, "name": kwargs.get("name", client_id)}
                self._handle_client_exception(e, "create_client", client_data)

            return {
                "client_id": client_id,
                "internal_client_id": internal_client_id,
                "realm": self.admin_adapter.connection.realm_name,
                "status": "created",
            }

        finally:
            # Always restore the original realm
            if realm and realm != original_realm:
                self.admin_adapter.connection.realm_name = original_realm

    @override
    def add_realm_roles_to_composite(self, composite_role_name: str, child_role_names: list[str]) -> None:
        """Add realm roles to a composite role.

        Args:
            composite_role_name: Name of the composite realm role
            child_role_names: List of child role names to add
        """
        try:
            child_roles = []
            for role_name in child_role_names:
                try:
                    role = self.admin_adapter.get_realm_role(role_name)
                    child_roles.append(role)
                except KeycloakGetError as e:
                    if e.response_code == 404:
                        logger.warning(f"Child role not found: {role_name}")
                        continue
                    raise

            if child_roles:
                self.admin_adapter.add_composite_realm_roles_to_role(role_name=composite_role_name, roles=child_roles)
                logger.info(f"Added {len(child_roles)} realm roles to composite role: {composite_role_name}")

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "add_realm_roles_to_composite")

    @override
    def add_client_roles_to_composite(
        self,
        composite_role_name: str,
        client_id: str,
        child_role_names: list[str],
    ) -> None:
        """Add client roles to a composite role.

        Args:
            composite_role_name: Name of the composite client role
            client_id: Client ID or client name
            child_role_names: List of child role names to add
        """
        try:
            internal_client_id = self.admin_adapter.get_client_id(client_id)
            if internal_client_id is None:
                raise ValueError("client_id resolved to None")

            child_roles = []
            for role_name in child_role_names:
                try:
                    # Keycloak admin adapter methods accept these types at runtime
                    role = self.admin_adapter.get_client_role(internal_client_id, role_name)
                    child_roles.append(role)
                except KeycloakGetError as e:
                    if e.response_code == 404:
                        logger.warning(f"Client role not found: {role_name}")
                        continue
                    raise

            if child_roles:
                if internal_client_id is None:
                    raise ValueError("Client ID not found")
                resolved_client_id: str = internal_client_id
                self.admin_adapter.add_composite_client_roles_to_role(
                    role_name=composite_role_name,
                    client_role_id=resolved_client_id,
                    roles=child_roles,
                )
                logger.info(f"Added {len(child_roles)} client roles to composite role: {composite_role_name}")

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "add_client_roles_to_composite")

    @override
    def get_composite_realm_roles(self, role_name: str) -> list[dict[str, Any]] | None:
        """Get composite roles for a realm role.

        Args:
            role_name: Name of the role

        Returns:
            List of composite roles
        """
        try:
            return self.admin_adapter.get_composite_realm_roles_of_role(role_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_composite_realm_roles")

    @override
    def get_organizations(self, query: dict | None = None) -> list[dict[str, Any]]:
        """Fetch all organizations, optionally filtered by query parameters."""
        try:
            return self.admin_adapter.get_organizations(query=query)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_organizations")

    @override
    def get_organization(self, organization_id: str) -> dict[str, Any]:
        """Get representation of the organization by ID."""
        try:
            return self.admin_adapter.get_organization(organization_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_organization")

    @override
    def create_organization(self, name: str, alias: str, **kwargs: Any) -> str | None:
        """Create a new organization. Name and alias must be unique. Returns org_id."""
        try:
            payload = {"name": name, "alias": alias}
            for key, value in kwargs.items():
                if key in ["name", "alias"]:
                    continue

                # Convert snake_case to camelCase
                camel_key = StringUtils.snake_to_camel_case(key)
                payload[camel_key] = value

            return self.admin_adapter.create_organization(payload=payload)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "create_organization")

    @override
    def update_organization(self, organization_id: str, **kwargs: Any) -> dict[str, Any]:
        """Update an existing organization. Kwargs are organization attributes (e.g. name, alias)."""
        try:
            payload = {}
            for key, value in kwargs.items():
                # Convert snake_case to camelCase
                camel_key = StringUtils.snake_to_camel_case(key)
                payload[camel_key] = value

            return self.admin_adapter.update_organization(organization_id=organization_id, payload=payload)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "update_organization")

    @override
    def delete_organization(self, organization_id: str) -> dict[str, Any]:
        """Delete an organization."""
        try:
            return self.admin_adapter.delete_organization(organization_id=organization_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "delete_organization")

    @override
    def get_user_organizations(self, user_id: str) -> list[dict[str, Any]]:
        """Get organizations by user id."""
        try:
            return self.admin_adapter.get_user_organizations(user_id=user_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_user_organizations")

    @override
    def get_organization_members(self, organization_id: str, query: dict | None = None) -> list[dict[str, Any]]:
        """Get members by organization id, optionally filtered by query parameters."""
        try:
            return self.admin_adapter.get_organization_members(organization_id=organization_id, query=query)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_organization_members")

    @override
    def get_organization_members_count(self, organization_id: str) -> int:
        """Get the number of members in the organization."""
        try:
            return self.admin_adapter.get_organization_members_count(organization_id=organization_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_organization_members_count")

    @override
    def organization_user_add(self, user_id: str, organization_id: str) -> bytes:
        """Add a user to an organization."""
        try:
            return self.admin_adapter.organization_user_add(user_id=user_id, organization_id=organization_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "organization_user_add")

    @override
    def organization_user_remove(self, user_id: str, organization_id: str) -> dict[str, Any]:
        """Remove a user from an organization."""
        try:
            return self.admin_adapter.organization_user_remove(user_id=user_id, organization_id=organization_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "organization_user_remove")

archipy.adapters.keycloak.adapters.KeycloakAdapter.configs instance-attribute

configs: KeycloakConfig = (
    KEYCLOAK
    if keycloak_configs is None
    else keycloak_configs
)

archipy.adapters.keycloak.adapters.KeycloakAdapter.admin_adapter property

admin_adapter: KeycloakAdmin

Get the admin adapter, refreshing it if necessary.

Returns:

Type	Description
KeycloakAdmin	KeycloakAdmin instance

Raises:

Type	Description
UnauthenticatedError	If admin client is not available due to authentication issues
UnavailableError	If Keycloak service is unavailable

archipy.adapters.keycloak.adapters.KeycloakAdapter.clear_all_caches

clear_all_caches() -> None

Clear all cached values.

Source code in archipy/adapters/keycloak/adapters.py

def clear_all_caches(self) -> None:
    """Clear all cached values."""
    for attr_name in dir(self):
        attr = getattr(self, attr_name)
        if hasattr(attr, "clear_cache"):
            attr.clear_cache()

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_public_key

get_public_key() -> PublicKeyType

Get the public key used to verify tokens.

Returns:

Type	Description
PublicKeyType	JWK key object used to verify signatures

Raises:

Type	Description
ServiceUnavailableError	If Keycloak service is unavailable
InternalError	If there's an internal error processing the public key

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=3600, maxsize=1)  # Cache for 1 hour, public key rarely changes
def get_public_key(self) -> PublicKeyType:
    """Get the public key used to verify tokens.

    Returns:
        JWK key object used to verify signatures

    Raises:
        ServiceUnavailableError: If Keycloak service is unavailable
        InternalError: If there's an internal error processing the public key
    """
    try:
        from jwcrypto import jwk

        keys_info = self._openid_adapter.public_key()
        key = f"-----BEGIN PUBLIC KEY-----\n{keys_info}\n-----END PUBLIC KEY-----"
        return jwk.JWK.from_pem(key.encode("utf-8"))
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_public_key")
    except Exception as e:
        raise InternalError(additional_data={"operation": "get_public_key", "error": str(e)}) from e

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_token

get_token(
    username: str, password: str
) -> KeycloakTokenType | None

Get a user token by username and password using the Resource Owner Password Credentials Grant.

Warning

This method uses the direct password grant flow, which is less secure and not recommended for user login in production environments. Instead, prefer the web-based OAuth 2.0 Authorization Code Flow (use get_token_from_code) for secure authentication. Use this method only for testing, administrative tasks, or specific service accounts where direct credential use is acceptable and properly secured.

Parameters:

Name	Type	Description	Default
username	str	User's username	required
password	str	User's password	required

Returns:

Type	Description
KeycloakTokenType | None	Token response containing access_token, refresh_token, etc.

Raises:

Type	Description
InvalidCredentialsError	If username or password is invalid
ServiceUnavailableError	If Keycloak service is unavailable

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_token(self, username: str, password: str) -> KeycloakTokenType | None:
    """Get a user token by username and password using the Resource Owner Password Credentials Grant.

    Warning:
        This method uses the direct password grant flow, which is less secure and not recommended
        for user login in production environments. Instead, prefer the web-based OAuth 2.0
        Authorization Code Flow (use `get_token_from_code`) for secure authentication.
        Use this method only for testing, administrative tasks, or specific service accounts
        where direct credential use is acceptable and properly secured.

    Args:
        username: User's username
        password: User's password

    Returns:
        Token response containing access_token, refresh_token, etc.

    Raises:
        InvalidCredentialsError: If username or password is invalid
        ServiceUnavailableError: If Keycloak service is unavailable
    """
    try:
        return self._openid_adapter.token(grant_type="password", username=username, password=password)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_token")

archipy.adapters.keycloak.adapters.KeycloakAdapter.refresh_token

refresh_token(
    refresh_token: str,
) -> KeycloakTokenType | None

Refresh an existing token using a refresh token.

Parameters:

Name	Type	Description	Default
refresh_token	str	Refresh token string	required

Returns:

Type	Description
KeycloakTokenType | None	New token response containing access_token, refresh_token, etc.

Raises:

Type	Description
InvalidTokenError	If refresh token is invalid or expired
ServiceUnavailableError	If Keycloak service is unavailable

Source code in archipy/adapters/keycloak/adapters.py

@override
def refresh_token(self, refresh_token: str) -> KeycloakTokenType | None:
    """Refresh an existing token using a refresh token.

    Args:
        refresh_token: Refresh token string

    Returns:
        New token response containing access_token, refresh_token, etc.

    Raises:
        InvalidTokenError: If refresh token is invalid or expired
        ServiceUnavailableError: If Keycloak service is unavailable
    """
    try:
        return self._openid_adapter.refresh_token(refresh_token)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "refresh_token")

archipy.adapters.keycloak.adapters.KeycloakAdapter.validate_token

validate_token(token: str) -> bool

Validate if a token is still valid.

Parameters:

Name	Type	Description	Default
token	str	Access token to validate	required

Returns:

Type	Description
bool	True if token is valid, False otherwise

Source code in archipy/adapters/keycloak/adapters.py

@override
def validate_token(self, token: str) -> bool:
    """Validate if a token is still valid.

    Args:
        token: Access token to validate

    Returns:
        True if token is valid, False otherwise
    """
    # Not caching validation results as tokens are time-sensitive
    try:
        # Let the underlying adapter handle key selection to align with expected types
        self._openid_adapter.decode_token(token)
    except Exception as e:
        logger.debug(f"Token validation failed: {e!s}")
        return False
    else:
        return True

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_userinfo

get_userinfo(token: str) -> KeycloakUserType | None

Get user information from a token.

Parameters:

Name	Type	Description	Default
token	str	Access token	required

Returns:

Type	Description
KeycloakUserType | None	User information

Raises:

Type	Description
ValueError	If getting user info fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_userinfo(self, token: str) -> KeycloakUserType | None:
    """Get user information from a token.

    Args:
        token: Access token

    Returns:
        User information

    Raises:
        ValueError: If getting user info fails
    """
    if not self.validate_token(token):
        raise InvalidTokenError()
    try:
        # _get_userinfo_cached returns KeycloakUserType (dict[str, Any])
        # The ttl_cache_decorator loses type info, but runtime behavior is correct
        # Access underlying function for proper typing
        cached_func = self._get_userinfo_cached
        underlying_func = getattr(cached_func, "__wrapped__", None)
        if underlying_func is not None:
            # Call underlying function directly for type checking
            result: KeycloakUserType = underlying_func(self, token)
        else:
            # Fallback to cached version if __wrapped__ not available
            result_raw = cached_func(token)
            if not isinstance(result_raw, dict):
                return None
            # Type assertion: result_raw is a dict, which matches KeycloakUserType
            # Convert to proper type by creating a new dict with explicit typing
            result: KeycloakUserType = {str(k): v for k, v in result_raw.items()}
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_userinfo")
        return None
    else:
        return result

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_user_by_id

get_user_by_id(user_id: str) -> KeycloakUserType | None

Get user details by user ID.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required

Returns:

Type	Description
KeycloakUserType | None	User details or None if not found

Raises:

Type	Description
ValueError	If getting user fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)  # Cache for 5 minutes
def get_user_by_id(self, user_id: str) -> KeycloakUserType | None:
    """Get user details by user ID.

    Args:
        user_id: User's ID

    Returns:
        User details or None if not found

    Raises:
        ValueError: If getting user fails
    """
    try:
        return self.admin_adapter.get_user(user_id)
    except KeycloakGetError as e:
        if e.response_code == 404:
            return None
        self._handle_keycloak_exception(e, "get_user_by_id")
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_user_by_id")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_user_by_username

get_user_by_username(
    username: str,
) -> KeycloakUserType | None

Get user details by username.

Parameters:

Name	Type	Description	Default
username	str	User's username	required

Returns:

Type	Description
KeycloakUserType | None	User details or None if not found

Raises:

Type	Description
ValueError	If query fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)  # Cache for 5 minutes
def get_user_by_username(self, username: str) -> KeycloakUserType | None:
    """Get user details by username.

    Args:
        username: User's username

    Returns:
        User details or None if not found

    Raises:
        ValueError: If query fails
    """
    try:
        users = self.admin_adapter.get_users({"username": username})
        return users[0] if users else None
    except KeycloakError as e:
        raise InternalError() from e

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_user_by_email

get_user_by_email(email: str) -> KeycloakUserType | None

Get user details by email.

Parameters:

Name	Type	Description	Default
email	str	User's email	required

Returns:

Type	Description
KeycloakUserType | None	User details or None if not found

Raises:

Type	Description
ValueError	If query fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)  # Cache for 5 minutes
def get_user_by_email(self, email: str) -> KeycloakUserType | None:
    """Get user details by email.

    Args:
        email: User's email

    Returns:
        User details or None if not found

    Raises:
        ValueError: If query fails
    """
    try:
        users = self.admin_adapter.get_users({"email": email})
        return users[0] if users else None
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_user_by_email")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_user_roles

get_user_roles(
    user_id: str,
) -> list[KeycloakRoleType] | None

Get roles assigned to a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required

Returns:

Type	Description
list[KeycloakRoleType] | None	List of roles

Raises:

Type	Description
ValueError	If getting roles fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)  # Cache for 5 minutes
def get_user_roles(self, user_id: str) -> list[KeycloakRoleType] | None:
    """Get roles assigned to a user.

    Args:
        user_id: User's ID

    Returns:
        List of roles

    Raises:
        ValueError: If getting roles fails
    """
    try:
        return self.admin_adapter.get_realm_roles_of_user(user_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_user_roles")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_client_roles_for_user

get_client_roles_for_user(
    user_id: str, client_id: str
) -> list[KeycloakRoleType]

Get client-specific roles assigned to a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
client_id	str	Client ID	required

Returns:

Type	Description
list[KeycloakRoleType]	List of client-specific roles

Raises:

Type	Description
ValueError	If getting roles fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)  # Cache for 5 minutes
def get_client_roles_for_user(self, user_id: str, client_id: str) -> list[KeycloakRoleType]:
    """Get client-specific roles assigned to a user.

    Args:
        user_id: User's ID
        client_id: Client ID

    Returns:
        List of client-specific roles

    Raises:
        ValueError: If getting roles fails
    """
    try:
        return self.admin_adapter.get_client_roles_of_user(user_id, client_id)
    except KeycloakError as e:
        raise InternalError() from e

archipy.adapters.keycloak.adapters.KeycloakAdapter.create_user

create_user(user_data: dict[str, Any]) -> str | None

Create a new user in Keycloak.

Parameters:

Name	Type	Description	Default
user_data	dict[str, Any]	User data including username, email, etc.	required

Returns:

Type	Description
str | None	ID of the created user

Raises:

Type	Description
ValueError	If creating user fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def create_user(self, user_data: dict[str, Any]) -> str | None:
    """Create a new user in Keycloak.

    Args:
        user_data: User data including username, email, etc.

    Returns:
        ID of the created user

    Raises:
        ValueError: If creating user fails
    """
    # This is a write operation, no caching needed
    try:
        user_id = self.admin_adapter.create_user(user_data)

        # Clear related caches
        self.clear_all_caches()

    except KeycloakError as e:
        self._handle_user_exception(e, "create_user", user_data)
    else:
        return user_id

archipy.adapters.keycloak.adapters.KeycloakAdapter.update_user

update_user(
    user_id: str, user_data: dict[str, Any]
) -> None

Update user details.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
user_data	dict[str, Any]	User data to update	required

Raises:

Type	Description
ValueError	If updating user fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def update_user(self, user_id: str, user_data: dict[str, Any]) -> None:
    """Update user details.

    Args:
        user_id: User's ID
        user_data: User data to update

    Raises:
        ValueError: If updating user fails
    """
    # This is a write operation, no caching needed
    try:
        self.admin_adapter.update_user(user_id, user_data)

        # Clear user-related caches
        self.clear_all_caches()

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "update_user")

archipy.adapters.keycloak.adapters.KeycloakAdapter.reset_password

reset_password(
    user_id: str, password: str, temporary: bool = False
) -> None

Reset a user's password.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
password	str	New password	required
temporary	bool	Whether the password is temporary and should be changed on next login	False

Raises:

Type	Description
ValueError	If password reset fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def reset_password(self, user_id: str, password: str, temporary: bool = False) -> None:
    """Reset a user's password.

    Args:
        user_id: User's ID
        password: New password
        temporary: Whether the password is temporary and should be changed on next login

    Raises:
        ValueError: If password reset fails
    """
    # This is a write operation, no caching needed
    try:
        self.admin_adapter.set_user_password(user_id, password, temporary)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "reset_password")

archipy.adapters.keycloak.adapters.KeycloakAdapter.assign_realm_role

assign_realm_role(user_id: str, role_name: str) -> None

Assign a realm role to a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
role_name	str	Role name to assign	required

Raises:

Type	Description
ValueError	If role assignment fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def assign_realm_role(self, user_id: str, role_name: str) -> None:
    """Assign a realm role to a user.

    Args:
        user_id: User's ID
        role_name: Role name to assign

    Raises:
        ValueError: If role assignment fails
    """
    # This is a write operation, no caching needed
    try:
        # Get role representation
        role = self.admin_adapter.get_realm_role(role_name)
        # Assign role to user
        self.admin_adapter.assign_realm_roles(user_id, [role])

        # Clear role-related caches
        if hasattr(self.get_user_roles, "clear_cache"):
            self.get_user_roles.clear_cache()

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "assign_realm_role")

archipy.adapters.keycloak.adapters.KeycloakAdapter.remove_realm_role

remove_realm_role(user_id: str, role_name: str) -> None

Remove a realm role from a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
role_name	str	Role name to remove	required

Raises:

Type	Description
ValueError	If role removal fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def remove_realm_role(self, user_id: str, role_name: str) -> None:
    """Remove a realm role from a user.

    Args:
        user_id: User's ID
        role_name: Role name to remove

    Raises:
        ValueError: If role removal fails
    """
    # This is a write operation, no caching needed
    try:
        # Get role representation
        role = self.admin_adapter.get_realm_role(role_name)
        # Remove role from user
        self.admin_adapter.delete_realm_roles_of_user(user_id, [role])

        # Clear role-related caches
        if hasattr(self.get_user_roles, "clear_cache"):
            self.get_user_roles.clear_cache()

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "remove_realm_role")

archipy.adapters.keycloak.adapters.KeycloakAdapter.assign_client_role

assign_client_role(
    user_id: str, client_id: str, role_name: str
) -> None

Assign a client-specific role to a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
client_id	str	Client ID	required
role_name	str	Role name to assign	required

Raises:

Type	Description
ValueError	If role assignment fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def assign_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
    """Assign a client-specific role to a user.

    Args:
        user_id: User's ID
        client_id: Client ID
        role_name: Role name to assign

    Raises:
        ValueError: If role assignment fails
    """
    # This is a write operation, no caching needed
    try:
        # Get client
        client = self.admin_adapter.get_client_id(client_id)
        if client is None:
            raise ValueError("client_id resolved to None")
        # Get role representation
        # Keycloak admin adapter methods accept these types at runtime
        role = self.admin_adapter.get_client_role(client, role_name)
        # Assign role to user
        self.admin_adapter.assign_client_role(user_id, client, [role])

        # Clear role-related caches
        if hasattr(self.get_client_roles_for_user, "clear_cache"):
            self.get_client_roles_for_user.clear_cache()

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "assign_client_role")

archipy.adapters.keycloak.adapters.KeycloakAdapter.create_realm_role

create_realm_role(
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None

Create a new realm role.

Parameters:

Name	Type	Description	Default
role_name	str	Role name	required
description	str | None	Optional role description	None
skip_exists	bool	Skip creation if realm role already exists	True

Returns:

Type	Description
dict[str, Any] | None	Created role details

Raises:

Type	Description
ValueError	If role creation fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def create_realm_role(
    self,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None:
    """Create a new realm role.

    Args:
        role_name: Role name
        description: Optional role description
        skip_exists: Skip creation if realm role already exists

    Returns:
        Created role details

    Raises:
        ValueError: If role creation fails
    """
    # This is a write operation, no caching needed
    try:
        role_data = {"name": role_name}
        if description:
            role_data["description"] = description

        self.admin_adapter.create_realm_role(role_data, skip_exists=skip_exists)

        # Clear realm roles cache
        if hasattr(self.get_realm_roles, "clear_cache"):
            self.get_realm_roles.clear_cache()

        return self.admin_adapter.get_realm_role(role_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "create_realm_role")

archipy.adapters.keycloak.adapters.KeycloakAdapter.create_client_role

create_client_role(
    client_id: str,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None

Create a new client role.

Parameters:

Name	Type	Description	Default
client_id	str	Client ID or client name	required
role_name	str	Role name	required
description	str | None	Optional role description	None
skip_exists	bool	Skip creation if client role already exists	True

Returns:

Type	Description
dict[str, Any] | None	Created role details

Raises:

Type	Description
ValueError	If role creation fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def create_client_role(
    self,
    client_id: str,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None:
    """Create a new client role.

    Args:
        client_id: Client ID or client name
        role_name: Role name
        description: Optional role description
        skip_exists: Skip creation if client role already exists

    Returns:
        Created role details

    Raises:
        ValueError: If role creation fails
    """
    # This is a write operation, no caching needed
    try:
        resolved_client_id = self.admin_adapter.get_client_id(client_id)
        if resolved_client_id is None:
            raise ValueError(f"Client ID not found: {client_id}")

        # Prepare role data
        role_data = {"name": role_name}
        if description:
            role_data["description"] = description

        # Create client role
        self.admin_adapter.create_client_role(resolved_client_id, role_data, skip_exists=skip_exists)

        # Clear related caches if they exist
        if hasattr(self.get_client_roles_for_user, "clear_cache"):
            self.get_client_roles_for_user.clear_cache()

        # Return created role
        return self.admin_adapter.get_client_role(resolved_client_id, role_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "create_client_role")

archipy.adapters.keycloak.adapters.KeycloakAdapter.delete_realm_role

delete_realm_role(role_name: str) -> None

Delete a realm role.

Parameters:

Name	Type	Description	Default
role_name	str	Role name to delete	required

Raises:

Type	Description
ValueError	If role deletion fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def delete_realm_role(self, role_name: str) -> None:
    """Delete a realm role.

    Args:
        role_name: Role name to delete

    Raises:
        ValueError: If role deletion fails
    """
    # This is a write operation, no caching needed
    try:
        self.admin_adapter.delete_realm_role(role_name)

        # Clear realm roles cache
        if hasattr(self.get_realm_roles, "clear_cache"):
            self.get_realm_roles.clear_cache()

        # We also need to clear user role caches since they might contain this role
        if hasattr(self.get_user_roles, "clear_cache"):
            self.get_user_roles.clear_cache()

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "delete_realm_role")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_service_account_id

get_service_account_id() -> str | None

Get service account user ID for the current client.

Returns:

Type	Description
str | None	Service account user ID

Raises:

Type	Description
ValueError	If getting service account fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=3600, maxsize=1)  # Cache for 1 hour
def get_service_account_id(self) -> str | None:
    """Get service account user ID for the current client.

    Returns:
        Service account user ID

    Raises:
        ValueError: If getting service account fails
    """
    try:
        client_id = self.get_client_id(self.configs.CLIENT_ID)
        return self.admin_adapter.get_client_service_account_user(str(client_id)).get("id")
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_service_account_id")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_well_known_config

get_well_known_config() -> dict[str, Any] | None

Get the well-known OpenID configuration.

Returns:

Type	Description
dict[str, Any] | None	OIDC configuration

Raises:

Type	Description
ValueError	If getting configuration fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=3600, maxsize=1)  # Cache for 1 hour
def get_well_known_config(self) -> dict[str, Any] | None:
    """Get the well-known OpenID configuration.

    Returns:
        OIDC configuration

    Raises:
        ValueError: If getting configuration fails
    """
    try:
        return self._openid_adapter.well_known()
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_well_known_config")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_certs

get_certs() -> dict[str, Any] | None

Get the JWT verification certificates.

Returns:

Type	Description
dict[str, Any] | None	Certificate information

Raises:

Type	Description
ValueError	If getting certificates fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=3600, maxsize=1)  # Cache for 1 hour
def get_certs(self) -> dict[str, Any] | None:
    """Get the JWT verification certificates.

    Returns:
        Certificate information

    Raises:
        ValueError: If getting certificates fails
    """
    try:
        return self._openid_adapter.certs()
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_certs")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_token_from_code

get_token_from_code(
    code: str, redirect_uri: str
) -> KeycloakTokenType | None

Exchange authorization code for token.

Parameters:

Name	Type	Description	Default
code	str	Authorization code	required
redirect_uri	str	Redirect URI used in authorization request	required

Returns:

Type	Description
KeycloakTokenType | None	Token response

Raises:

Type	Description
ValueError	If token exchange fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_token_from_code(self, code: str, redirect_uri: str) -> KeycloakTokenType | None:
    """Exchange authorization code for token.

    Args:
        code: Authorization code
        redirect_uri: Redirect URI used in authorization request

    Returns:
        Token response

    Raises:
        ValueError: If token exchange fails
    """
    # Authorization codes can only be used once, don't cache
    try:
        return self._openid_adapter.token(grant_type="authorization_code", code=code, redirect_uri=redirect_uri)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_token_from_code")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_client_credentials_token

get_client_credentials_token() -> KeycloakTokenType | None

Get token using client credentials.

Returns:

Type	Description
KeycloakTokenType | None	Token response

Raises:

Type	Description
ValueError	If token acquisition fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_client_credentials_token(self) -> KeycloakTokenType | None:
    """Get token using client credentials.

    Returns:
        Token response

    Raises:
        ValueError: If token acquisition fails
    """
    # Tokens are time-sensitive, don't cache
    try:
        return self._openid_adapter.token(grant_type="client_credentials")
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_client_credentials_token")

archipy.adapters.keycloak.adapters.KeycloakAdapter.search_users

search_users(
    query: str, max_results: int = 100
) -> list[KeycloakUserType] | None

Search for users by username, email, or name.

Parameters:

Name	Type	Description	Default
query	str	Search query	required
max_results	int	Maximum number of results to return	100

Returns:

Type	Description
list[KeycloakUserType] | None	List of matching users

Raises:

Type	Description
ValueError	If search fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=30, maxsize=50)  # Cache for 30 seconds with limited entries
def search_users(self, query: str, max_results: int = 100) -> list[KeycloakUserType] | None:
    """Search for users by username, email, or name.

    Args:
        query: Search query
        max_results: Maximum number of results to return

    Returns:
        List of matching users

    Raises:
        ValueError: If search fails
    """
    try:
        # Try searching by different fields
        users = []

        # Search by username
        users.extend(self.admin_adapter.get_users({"username": query, "max": max_results}))

        # Search by email if no results or incomplete results
        if len(users) < max_results:
            remaining = max_results - len(users)
            email_users = self.admin_adapter.get_users({"email": query, "max": remaining})
            # Filter out duplicates
            user_ids = {user["id"] for user in users}
            users.extend([user for user in email_users if user["id"] not in user_ids])

        # Search by firstName if no results or incomplete results
        if len(users) < max_results:
            remaining = max_results - len(users)
            first_name_users = self.admin_adapter.get_users({"firstName": query, "max": remaining})
            # Filter out duplicates
            user_ids = {user["id"] for user in users}
            users.extend([user for user in first_name_users if user["id"] not in user_ids])

        # Search by lastName if no results or incomplete results
        if len(users) < max_results:
            remaining = max_results - len(users)
            last_name_users = self.admin_adapter.get_users({"lastName": query, "max": remaining})
            # Filter out duplicates
            user_ids = {user["id"] for user in users}
            users.extend([user for user in last_name_users if user["id"] not in user_ids])

        return users[:max_results]
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "search_users")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_client_secret

get_client_secret(client_id: str) -> str | None

Get client secret.

Parameters:

Name	Type	Description	Default
client_id	str	Client ID	required

Returns:

Type	Description
str | None	Client secret

Raises:

Type	Description
ValueError	If getting secret fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=3600, maxsize=50)  # Cache for 1 hour
def get_client_secret(self, client_id: str) -> str | None:
    """Get client secret.

    Args:
        client_id: Client ID

    Returns:
        Client secret

    Raises:
        ValueError: If getting secret fails
    """
    try:
        client = self.admin_adapter.get_client(client_id)
        return client.get("secret", "")
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_client_secret")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_client_id

get_client_id(client_name: str) -> str | None

Get client ID by client name.

Parameters:

Name	Type	Description	Default
client_name	str	Name of the client	required

Returns:

Type	Description
str | None	Client ID

Raises:

Type	Description
ValueError	If client not found

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=3600, maxsize=50)  # Cache for 1 hour
def get_client_id(self, client_name: str) -> str | None:
    """Get client ID by client name.

    Args:
        client_name: Name of the client

    Returns:
        Client ID

    Raises:
        ValueError: If client not found
    """
    try:
        return self.admin_adapter.get_client_id(client_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_client_id")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_realm_roles

get_realm_roles() -> list[dict[str, Any]] | None

Get all realm roles.

Returns:

Type	Description
list[dict[str, Any]] | None	List of realm roles

Raises:

Type	Description
ValueError	If getting roles fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=1)  # Cache for 5 minutes
def get_realm_roles(self) -> list[dict[str, Any]] | None:
    """Get all realm roles.

    Returns:
        List of realm roles

    Raises:
        ValueError: If getting roles fails
    """
    try:
        return self.admin_adapter.get_realm_roles()
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_realm_roles")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_realm_role

get_realm_role(role_name: str) -> dict | None

Get realm role.

Parameters:

Name	Type	Description	Default
role_name	str	Role name	required

Returns: A realm role

Raises:

Type	Description
ValueError	If getting role fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=1)  # Cache for 5 minutes
def get_realm_role(self, role_name: str) -> dict | None:
    """Get realm role.

    Args:
        role_name: Role name
    Returns:
        A realm role

    Raises:
        ValueError: If getting role fails
    """
    try:
        return self.admin_adapter.get_realm_role(role_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_realm_role")

archipy.adapters.keycloak.adapters.KeycloakAdapter.remove_client_role

remove_client_role(
    user_id: str, client_id: str, role_name: str
) -> None

Remove a client-specific role from a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
client_id	str	Client ID	required
role_name	str	Role name to remove	required

Raises:

Type	Description
ValueError	If role removal fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def remove_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
    """Remove a client-specific role from a user.

    Args:
        user_id: User's ID
        client_id: Client ID
        role_name: Role name to remove

    Raises:
        ValueError: If role removal fails
    """
    try:
        client = self.admin_adapter.get_client_id(client_id)
        if client is None:
            raise ValueError("client_id resolved to None")
        # Keycloak admin adapter methods accept these types at runtime
        role = self.admin_adapter.get_client_role(client, role_name)
        self.admin_adapter.delete_client_roles_of_user(user_id, client, [role])

        if hasattr(self.get_client_roles_for_user, "clear_cache"):
            self.get_client_roles_for_user.clear_cache()
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "remove_client_role")

archipy.adapters.keycloak.adapters.KeycloakAdapter.clear_user_sessions

clear_user_sessions(user_id: str) -> None

Clear all sessions for a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required

Raises:

Type	Description
ValueError	If clearing sessions fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def clear_user_sessions(self, user_id: str) -> None:
    """Clear all sessions for a user.

    Args:
        user_id: User's ID

    Raises:
        ValueError: If clearing sessions fails
    """
    try:
        self.admin_adapter.user_logout(user_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "clear_user_sessions")

archipy.adapters.keycloak.adapters.KeycloakAdapter.logout

logout(refresh_token: str) -> None

Logout user by invalidating their refresh token.

Parameters:

Name	Type	Description	Default
refresh_token	str	Refresh token to invalidate	required

Raises:

Type	Description
ValueError	If logout fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def logout(self, refresh_token: str) -> None:
    """Logout user by invalidating their refresh token.

    Args:
        refresh_token: Refresh token to invalidate

    Raises:
        ValueError: If logout fails
    """
    try:
        self._openid_adapter.logout(refresh_token)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "logout")

archipy.adapters.keycloak.adapters.KeycloakAdapter.introspect_token

introspect_token(token: str) -> dict[str, Any] | None

Introspect token to get detailed information about it.

Parameters:

Name	Type	Description	Default
token	str	Access token	required

Returns:

Type	Description
dict[str, Any] | None	Token introspection details

Raises:

Type	Description
ValueError	If token introspection fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def introspect_token(self, token: str) -> dict[str, Any] | None:
    """Introspect token to get detailed information about it.

    Args:
        token: Access token

    Returns:
        Token introspection details

    Raises:
        ValueError: If token introspection fails
    """
    try:
        return self._openid_adapter.introspect(token)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "introspect_token")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_token_info

get_token_info(token: str) -> dict[str, Any] | None

Decode token to get its claims.

Parameters:

Name	Type	Description	Default
token	str	Access token	required

Returns:

Type	Description
dict[str, Any] | None	Dictionary of token claims

Raises:

Type	Description
ValueError	If token decoding fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_token_info(self, token: str) -> dict[str, Any] | None:
    """Decode token to get its claims.

    Args:
        token: Access token

    Returns:
        Dictionary of token claims

    Raises:
        ValueError: If token decoding fails
    """
    try:
        # Let the underlying adapter handle key selection to align with expected types
        return self._openid_adapter.decode_token(token)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_token_info")

archipy.adapters.keycloak.adapters.KeycloakAdapter.delete_user

delete_user(user_id: str) -> None

Delete a user from Keycloak by their ID.

Parameters:

Name	Type	Description	Default
user_id	str	The ID of the user to delete	required

Raises:

Type	Description
ValueError	If the deletion fails

Source code in archipy/adapters/keycloak/adapters.py

@override
def delete_user(self, user_id: str) -> None:
    """Delete a user from Keycloak by their ID.

    Args:
        user_id: The ID of the user to delete

    Raises:
        ValueError: If the deletion fails
    """
    try:
        self.admin_adapter.delete_user(user_id=user_id)

        if hasattr(self.get_user_by_username, "clear_cache"):
            self.get_user_by_username.clear_cache()

        logger.info(f"Successfully deleted user with ID {user_id}")
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "delete_user")

archipy.adapters.keycloak.adapters.KeycloakAdapter.has_role

has_role(token: str, role_name: str) -> bool

Check if a user has a specific role.

Parameters:

Name	Type	Description	Default
token	str	Access token	required
role_name	str	Role name to check	required

Returns:

Type	Description
bool	True if user has the role, False otherwise

Source code in archipy/adapters/keycloak/adapters.py

@override
def has_role(self, token: str, role_name: str) -> bool:
    """Check if a user has a specific role.

    Args:
        token: Access token
        role_name: Role name to check

    Returns:
        True if user has the role, False otherwise
    """
    # Not caching this result as token validation is time-sensitive
    try:
        user_info = self.get_userinfo(token)
        if not user_info:
            return False

        # Check realm roles
        realm_access = user_info.get("realm_access", {})
        roles = realm_access.get("roles", [])
        if role_name in roles:
            return True

        # Check client roles
        resource_access = user_info.get("resource_access", {})
        client_roles = resource_access.get(self.configs.CLIENT_ID, {}).get("roles", [])
        if role_name in client_roles:
            return True

    except Exception as e:
        logger.debug(f"Role check failed: {e!s}")
        return False
    else:
        return False

archipy.adapters.keycloak.adapters.KeycloakAdapter.has_any_of_roles

has_any_of_roles(
    token: str, role_names: frozenset[str]
) -> bool

Check if a user has any of the specified roles.

Parameters:

Name	Type	Description	Default
token	str	Access token	required
role_names	frozenset[str]	Set of role names to check	required

Returns:

Type	Description
bool	True if user has any of the roles, False otherwise

Source code in archipy/adapters/keycloak/adapters.py

@override
def has_any_of_roles(self, token: str, role_names: frozenset[str]) -> bool:
    """Check if a user has any of the specified roles.

    Args:
        token: Access token
        role_names: Set of role names to check

    Returns:
        True if user has any of the roles, False otherwise
    """
    try:
        user_info = self.get_userinfo(token)
        if not user_info:
            return False

        # Check realm roles first
        realm_access = user_info.get("realm_access", {})
        realm_roles = set(realm_access.get("roles", []))
        if role_names.intersection(realm_roles):
            return True

        # Check roles for the configured client
        resource_access = user_info.get("resource_access", {})
        client_roles = set(resource_access.get(self.configs.CLIENT_ID, {}).get("roles", []))
        if role_names.intersection(client_roles):
            return True

    except Exception as e:
        logger.debug(f"Role check failed: {e!s}")
        return False
    else:
        return False

archipy.adapters.keycloak.adapters.KeycloakAdapter.has_all_roles

has_all_roles(
    token: str, role_names: frozenset[str]
) -> bool

Check if a user has all the specified roles.

Parameters:

Name	Type	Description	Default
token	str	Access token	required
role_names	frozenset[str]	Set of role names to check	required

Returns:

Type	Description
bool	True if user has all the roles, False otherwise

Source code in archipy/adapters/keycloak/adapters.py

@override
def has_all_roles(self, token: str, role_names: frozenset[str]) -> bool:
    """Check if a user has all the specified roles.

    Args:
        token: Access token
        role_names: Set of role names to check

    Returns:
        True if user has all the roles, False otherwise
    """
    try:
        user_info = self.get_userinfo(token)
        if not user_info:
            return False

        # Get all user roles
        all_roles = set()

        # Add realm roles
        realm_access = user_info.get("realm_access", {})
        all_roles.update(realm_access.get("roles", []))

        # Add client roles
        resource_access = user_info.get("resource_access", {})
        client_roles = resource_access.get(self.configs.CLIENT_ID, {}).get("roles", [])
        all_roles.update(client_roles)

        # Check if all required roles are present
        return role_names.issubset(all_roles)

    except Exception as e:
        logger.debug(f"All roles check failed: {e!s}")
        return False

archipy.adapters.keycloak.adapters.KeycloakAdapter.check_permissions

check_permissions(
    token: str, resource: str, scope: str
) -> bool

Check if a user has permission to access a resource with the specified scope.

Parameters:

Name	Type	Description	Default
token	str	Access token	required
resource	str	Resource name	required
scope	str	Permission scope	required

Returns:

Type	Description
bool	True if permission granted, False otherwise

Source code in archipy/adapters/keycloak/adapters.py

@override
def check_permissions(self, token: str, resource: str, scope: str) -> bool:
    """Check if a user has permission to access a resource with the specified scope.

    Args:
        token: Access token
        resource: Resource name
        scope: Permission scope

    Returns:
        True if permission granted, False otherwise
    """
    try:
        # Use UMA permissions endpoint to check specific resource and scope
        permissions = self._openid_adapter.uma_permissions(token, permissions=f"{resource}#{scope}")

        # Check if the response indicates permission is granted
        if not permissions or not isinstance(permissions, list):
            logger.debug("No permissions returned or invalid response format")
            return False

        # Look for the specific permission in the response
        for perm in permissions:
            if perm.get("rsname") == resource and scope in perm.get("scopes", []):
                return True

    except KeycloakError as e:
        logger.debug(f"Permission check failed with Keycloak error: {e!s}")
        return False
    except Exception as e:
        logger.debug(f"Permission check failed with unexpected error: {e!s}")
        return False
    else:
        return False

archipy.adapters.keycloak.adapters.KeycloakAdapter.create_realm

create_realm(
    realm_name: str, skip_exists: bool = True, **kwargs: Any
) -> dict[str, Any] | None

Create a Keycloak realm with minimum required fields and optional additional config.

Parameters:

Name	Type	Description	Default
realm_name	str	The realm identifier (required)	required
skip_exists	bool	Skip creation if realm already exists	True
kwargs	Any	Additional optional configurations for the realm	{}

Returns:

Type	Description
dict[str, Any] | None	Realm details

Source code in archipy/adapters/keycloak/adapters.py

@override
def create_realm(self, realm_name: str, skip_exists: bool = True, **kwargs: Any) -> dict[str, Any] | None:
    """Create a Keycloak realm with minimum required fields and optional additional config.

    Args:
        realm_name: The realm identifier (required)
        skip_exists: Skip creation if realm already exists
        kwargs: Additional optional configurations for the realm

    Returns:
        Realm details
    """
    payload = {
        "realm": realm_name,
        "enabled": kwargs.get("enabled", True),
        "displayName": kwargs.get("display_name", realm_name),
    }

    # Add any additional parameters from kwargs
    for key, value in kwargs.items():
        # Skip display_name as it's already handled
        if key == "display_name":
            continue

        # Convert Python snake_case to Keycloak camelCase
        camel_key = StringUtils.snake_to_camel_case(key)
        payload[camel_key] = value

    try:
        self.admin_adapter.create_realm(payload=payload, skip_exists=skip_exists)
    except KeycloakError as e:
        logger.debug(f"Failed to create realm: {e!s}")

        # Handle realm already exists with skip_exists option
        if skip_exists:
            error_message = self._extract_error_message(e).lower()
            if "already exists" in error_message and "realm" in error_message:
                return {"realm": realm_name, "status": "already_exists", "config": payload}

        # Use the mixin to handle realm-specific errors
        self._handle_realm_exception(e, "create_realm", realm_name)
    else:
        return {"realm": realm_name, "status": "created", "config": payload}

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_realm

get_realm(realm_name: str) -> dict[str, Any] | None

Get realm details by realm name.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required

Returns:

Type	Description
dict[str, Any] | None	Realm details

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_realm(self, realm_name: str) -> dict[str, Any] | None:
    """Get realm details by realm name.

    Args:
        realm_name: Name of the realm

    Returns:
        Realm details
    """
    try:
        return self.admin_adapter.get_realm(realm_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_realm")

archipy.adapters.keycloak.adapters.KeycloakAdapter.update_realm

update_realm(
    realm_name: str, **kwargs: Any
) -> dict[str, Any] | None

Update a realm. Kwargs are RealmRepresentation.

Parameters:

Name	Type	Description	Default
realm_name	str	Realm name (not the realm id).	required
**kwargs	Any	RealmRepresentation attributes to update (e.g. displayName).	{}

Returns:

Type	Description
dict[str, Any] | None	Response from Keycloak, or None on error (handled via exception).

Source code in archipy/adapters/keycloak/adapters.py

@override
def update_realm(self, realm_name: str, **kwargs: Any) -> dict[str, Any] | None:
    """Update a realm. Kwargs are RealmRepresentation.

    Args:
        realm_name: Realm name (not the realm id).
        **kwargs: RealmRepresentation attributes to update (e.g. displayName).

    Returns:
        Response from Keycloak, or None on error (handled via exception).
    """
    try:
        return self.admin_adapter.update_realm(realm_name, dict(kwargs))
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "update_realm")

archipy.adapters.keycloak.adapters.KeycloakAdapter.create_client

create_client(
    client_id: str,
    realm: str | None = None,
    skip_exists: bool = True,
    **kwargs: Any,
) -> dict[str, Any] | None

Create a Keycloak client with minimum required fields and optional additional config.

Parameters:

Name	Type	Description	Default
client_id	str	The client identifier (required)	required
realm	str | None	Target realm name (uses the current realm in KeycloakAdmin if not specified)	None
skip_exists	bool	Skip creation if client already exists	True
kwargs	Any	Additional optional configurations for the client	{}

Returns:

Type	Description
dict[str, Any] | None	Client details

Source code in archipy/adapters/keycloak/adapters.py

@override
def create_client(
    self,
    client_id: str,
    realm: str | None = None,
    skip_exists: bool = True,
    **kwargs: Any,
) -> dict[str, Any] | None:
    """Create a Keycloak client with minimum required fields and optional additional config.

    Args:
        client_id: The client identifier (required)
        realm: Target realm name (uses the current realm in KeycloakAdmin if not specified)
        skip_exists: Skip creation if client already exists
        kwargs: Additional optional configurations for the client

    Returns:
        Client details
    """
    original_realm = self.admin_adapter.connection.realm_name

    try:
        # Set the target realm if provided
        if realm and realm != original_realm:
            self.admin_adapter.connection.realm_name = realm

        public_client = kwargs.get("public_client", False)

        # Prepare the minimal client payload
        payload = {
            "clientId": client_id,
            "enabled": kwargs.get("enabled", True),
            "protocol": kwargs.get("protocol", "openid-connect"),
            "name": kwargs.get("name", client_id),
            "publicClient": public_client,
        }

        # Enable service accounts for confidential clients by default
        if not public_client:
            payload["serviceAccountsEnabled"] = kwargs.get("service_account_enabled", True)
            payload["clientAuthenticatorType"] = "client-secret"

        for key, value in kwargs.items():
            if key in ["enabled", "protocol", "name", "public_client", "service_account_enabled"]:
                continue

            # Convert snake_case to camelCase
            camel_key = StringUtils.snake_to_camel_case(key)
            payload[camel_key] = value

        internal_client_id = None
        try:
            internal_client_id = self.admin_adapter.create_client(payload, skip_exists=skip_exists)
        except KeycloakError as e:
            logger.debug(f"Failed to create client: {e!s}")

            # Handle client already exists with skip_exists option
            if skip_exists:
                error_message = self._extract_error_message(e).lower()
                if "already exists" in error_message and "client" in error_message:
                    return {
                        "client_id": client_id,
                        "status": "already_exists",
                        "realm": self.admin_adapter.connection.realm_name,
                    }

            # Use the mixin to handle client-specific errors
            client_data = {"clientId": client_id, "name": kwargs.get("name", client_id)}
            self._handle_client_exception(e, "create_client", client_data)

        return {
            "client_id": client_id,
            "internal_client_id": internal_client_id,
            "realm": self.admin_adapter.connection.realm_name,
            "status": "created",
        }

    finally:
        # Always restore the original realm
        if realm and realm != original_realm:
            self.admin_adapter.connection.realm_name = original_realm

archipy.adapters.keycloak.adapters.KeycloakAdapter.add_realm_roles_to_composite

add_realm_roles_to_composite(
    composite_role_name: str, child_role_names: list[str]
) -> None

Add realm roles to a composite role.

Parameters:

Name	Type	Description	Default
composite_role_name	str	Name of the composite realm role	required
child_role_names	list[str]	List of child role names to add	required

Source code in archipy/adapters/keycloak/adapters.py

@override
def add_realm_roles_to_composite(self, composite_role_name: str, child_role_names: list[str]) -> None:
    """Add realm roles to a composite role.

    Args:
        composite_role_name: Name of the composite realm role
        child_role_names: List of child role names to add
    """
    try:
        child_roles = []
        for role_name in child_role_names:
            try:
                role = self.admin_adapter.get_realm_role(role_name)
                child_roles.append(role)
            except KeycloakGetError as e:
                if e.response_code == 404:
                    logger.warning(f"Child role not found: {role_name}")
                    continue
                raise

        if child_roles:
            self.admin_adapter.add_composite_realm_roles_to_role(role_name=composite_role_name, roles=child_roles)
            logger.info(f"Added {len(child_roles)} realm roles to composite role: {composite_role_name}")

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "add_realm_roles_to_composite")

archipy.adapters.keycloak.adapters.KeycloakAdapter.add_client_roles_to_composite

add_client_roles_to_composite(
    composite_role_name: str,
    client_id: str,
    child_role_names: list[str],
) -> None

Add client roles to a composite role.

Parameters:

Name	Type	Description	Default
composite_role_name	str	Name of the composite client role	required
client_id	str	Client ID or client name	required
child_role_names	list[str]	List of child role names to add	required

Source code in archipy/adapters/keycloak/adapters.py

@override
def add_client_roles_to_composite(
    self,
    composite_role_name: str,
    client_id: str,
    child_role_names: list[str],
) -> None:
    """Add client roles to a composite role.

    Args:
        composite_role_name: Name of the composite client role
        client_id: Client ID or client name
        child_role_names: List of child role names to add
    """
    try:
        internal_client_id = self.admin_adapter.get_client_id(client_id)
        if internal_client_id is None:
            raise ValueError("client_id resolved to None")

        child_roles = []
        for role_name in child_role_names:
            try:
                # Keycloak admin adapter methods accept these types at runtime
                role = self.admin_adapter.get_client_role(internal_client_id, role_name)
                child_roles.append(role)
            except KeycloakGetError as e:
                if e.response_code == 404:
                    logger.warning(f"Client role not found: {role_name}")
                    continue
                raise

        if child_roles:
            if internal_client_id is None:
                raise ValueError("Client ID not found")
            resolved_client_id: str = internal_client_id
            self.admin_adapter.add_composite_client_roles_to_role(
                role_name=composite_role_name,
                client_role_id=resolved_client_id,
                roles=child_roles,
            )
            logger.info(f"Added {len(child_roles)} client roles to composite role: {composite_role_name}")

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "add_client_roles_to_composite")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_composite_realm_roles

get_composite_realm_roles(
    role_name: str,
) -> list[dict[str, Any]] | None

Get composite roles for a realm role.

Parameters:

Name	Type	Description	Default
role_name	str	Name of the role	required

Returns:

Type	Description
list[dict[str, Any]] | None	List of composite roles

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_composite_realm_roles(self, role_name: str) -> list[dict[str, Any]] | None:
    """Get composite roles for a realm role.

    Args:
        role_name: Name of the role

    Returns:
        List of composite roles
    """
    try:
        return self.admin_adapter.get_composite_realm_roles_of_role(role_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_composite_realm_roles")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_organizations

get_organizations(
    query: dict | None = None,
) -> list[dict[str, Any]]

Fetch all organizations, optionally filtered by query parameters.

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_organizations(self, query: dict | None = None) -> list[dict[str, Any]]:
    """Fetch all organizations, optionally filtered by query parameters."""
    try:
        return self.admin_adapter.get_organizations(query=query)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_organizations")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_organization

get_organization(organization_id: str) -> dict[str, Any]

Get representation of the organization by ID.

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_organization(self, organization_id: str) -> dict[str, Any]:
    """Get representation of the organization by ID."""
    try:
        return self.admin_adapter.get_organization(organization_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_organization")

archipy.adapters.keycloak.adapters.KeycloakAdapter.create_organization

create_organization(
    name: str, alias: str, **kwargs: Any
) -> str | None

Create a new organization. Name and alias must be unique. Returns org_id.

Source code in archipy/adapters/keycloak/adapters.py

@override
def create_organization(self, name: str, alias: str, **kwargs: Any) -> str | None:
    """Create a new organization. Name and alias must be unique. Returns org_id."""
    try:
        payload = {"name": name, "alias": alias}
        for key, value in kwargs.items():
            if key in ["name", "alias"]:
                continue

            # Convert snake_case to camelCase
            camel_key = StringUtils.snake_to_camel_case(key)
            payload[camel_key] = value

        return self.admin_adapter.create_organization(payload=payload)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "create_organization")

archipy.adapters.keycloak.adapters.KeycloakAdapter.update_organization

update_organization(
    organization_id: str, **kwargs: Any
) -> dict[str, Any]

Update an existing organization. Kwargs are organization attributes (e.g. name, alias).

Source code in archipy/adapters/keycloak/adapters.py

@override
def update_organization(self, organization_id: str, **kwargs: Any) -> dict[str, Any]:
    """Update an existing organization. Kwargs are organization attributes (e.g. name, alias)."""
    try:
        payload = {}
        for key, value in kwargs.items():
            # Convert snake_case to camelCase
            camel_key = StringUtils.snake_to_camel_case(key)
            payload[camel_key] = value

        return self.admin_adapter.update_organization(organization_id=organization_id, payload=payload)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "update_organization")

archipy.adapters.keycloak.adapters.KeycloakAdapter.delete_organization

delete_organization(organization_id: str) -> dict[str, Any]

Delete an organization.

Source code in archipy/adapters/keycloak/adapters.py

@override
def delete_organization(self, organization_id: str) -> dict[str, Any]:
    """Delete an organization."""
    try:
        return self.admin_adapter.delete_organization(organization_id=organization_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "delete_organization")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_user_organizations

get_user_organizations(
    user_id: str,
) -> list[dict[str, Any]]

Get organizations by user id.

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_user_organizations(self, user_id: str) -> list[dict[str, Any]]:
    """Get organizations by user id."""
    try:
        return self.admin_adapter.get_user_organizations(user_id=user_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_user_organizations")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_organization_members

get_organization_members(
    organization_id: str, query: dict | None = None
) -> list[dict[str, Any]]

Get members by organization id, optionally filtered by query parameters.

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_organization_members(self, organization_id: str, query: dict | None = None) -> list[dict[str, Any]]:
    """Get members by organization id, optionally filtered by query parameters."""
    try:
        return self.admin_adapter.get_organization_members(organization_id=organization_id, query=query)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_organization_members")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_organization_members_count

get_organization_members_count(organization_id: str) -> int

Get the number of members in the organization.

Source code in archipy/adapters/keycloak/adapters.py

@override
def get_organization_members_count(self, organization_id: str) -> int:
    """Get the number of members in the organization."""
    try:
        return self.admin_adapter.get_organization_members_count(organization_id=organization_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_organization_members_count")

archipy.adapters.keycloak.adapters.KeycloakAdapter.organization_user_add

organization_user_add(
    user_id: str, organization_id: str
) -> bytes

Add a user to an organization.

Source code in archipy/adapters/keycloak/adapters.py

@override
def organization_user_add(self, user_id: str, organization_id: str) -> bytes:
    """Add a user to an organization."""
    try:
        return self.admin_adapter.organization_user_add(user_id=user_id, organization_id=organization_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "organization_user_add")

archipy.adapters.keycloak.adapters.KeycloakAdapter.organization_user_remove

organization_user_remove(
    user_id: str, organization_id: str
) -> dict[str, Any]

Remove a user from an organization.

Source code in archipy/adapters/keycloak/adapters.py

@override
def organization_user_remove(self, user_id: str, organization_id: str) -> dict[str, Any]:
    """Remove a user from an organization."""
    try:
        return self.admin_adapter.organization_user_remove(user_id=user_id, organization_id=organization_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "organization_user_remove")

archipy.adapters.keycloak.adapters.KeycloakAdapter.get_organization_idps abstractmethod

get_organization_idps(
    organization_id: str,
) -> list[dict[str, Any]]

Get IDPs by organization id.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
def get_organization_idps(self, organization_id: str) -> list[dict[str, Any]]:
    """Get IDPs by organization id."""
    raise NotImplementedError

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter

Bases: AsyncKeycloakPort, KeycloakExceptionHandlerMixin

Concrete implementation of the KeycloakPort interface using python-keycloak library.

This implementation includes TTL caching for appropriate operations to improve performance while ensuring cache entries expire after a configured time to prevent stale data.

Source code in archipy/adapters/keycloak/adapters.py

class AsyncKeycloakAdapter(AsyncKeycloakPort, KeycloakExceptionHandlerMixin):
    """Concrete implementation of the KeycloakPort interface using python-keycloak library.

    This implementation includes TTL caching for appropriate operations to improve performance
    while ensuring cache entries expire after a configured time to prevent stale data.
    """

    def __init__(self, keycloak_configs: KeycloakConfig | None = None) -> None:
        """Initialize KeycloakAdapter with configuration.

        Args:
            keycloak_configs: Optional Keycloak configuration. If None, global config is used.
        """
        self.configs: KeycloakConfig = (
            BaseConfig.global_config().KEYCLOAK if keycloak_configs is None else keycloak_configs
        )

        # Initialize the OpenID client for authentication
        self.openid_adapter = self._get_openid_client(self.configs)

        # Cache for admin client to avoid unnecessary re-authentication
        self._admin_adapter: KeycloakAdmin | None = None
        self._admin_token_expiry: float = 0.0

        # Initialize admin client if admin mode is enabled and credentials are provided
        if self.configs.IS_ADMIN_MODE_ENABLED and (
            self.configs.CLIENT_SECRET_KEY or (self.configs.ADMIN_USERNAME and self.configs.ADMIN_PASSWORD)
        ):
            self._initialize_admin_client()

    def clear_all_caches(self) -> None:
        """Clear all cached values."""
        for attr_name in dir(self):
            attr = getattr(self, attr_name)
            if hasattr(attr, "cache_clear"):
                attr.cache_clear()

    @staticmethod
    def _get_openid_client(configs: KeycloakConfig) -> KeycloakOpenID:
        """Create and configure a KeycloakOpenID instance.

        Args:
            configs: Keycloak configuration

        Returns:
            Configured KeycloakOpenID client
        """
        server_url = configs.SERVER_URL
        client_id = configs.CLIENT_ID
        if not server_url or not client_id:
            raise ValueError("SERVER_URL and CLIENT_ID must be provided")
        return KeycloakOpenID(
            server_url=server_url,
            client_id=client_id,
            realm_name=configs.REALM_NAME,
            client_secret_key=configs.CLIENT_SECRET_KEY,
            verify=configs.VERIFY_SSL,
            timeout=configs.TIMEOUT,
        )

    def _initialize_admin_client(self) -> None:
        """Initialize or refresh the admin client."""
        try:
            # Check if admin credentials are available
            if self.configs.ADMIN_USERNAME and self.configs.ADMIN_PASSWORD:
                # Create admin client using admin credentials
                self._admin_adapter = KeycloakAdmin(
                    server_url=self.configs.SERVER_URL,
                    username=self.configs.ADMIN_USERNAME,
                    password=self.configs.ADMIN_PASSWORD,
                    realm_name=self.configs.REALM_NAME,
                    user_realm_name=self.configs.ADMIN_REALM_NAME,
                    verify=self.configs.VERIFY_SSL,
                    timeout=self.configs.TIMEOUT,
                )
                # Since we're using direct credentials, set a long expiry time
                self._admin_token_expiry = time.time() + 3600  # 1 hour
                logger.debug("Admin client initialized with admin credentials")
            elif self.configs.CLIENT_SECRET_KEY:
                # Get token using client credentials
                token = self.openid_adapter.token(grant_type="client_credentials")

                # Set token expiry time (current time + expires_in - buffer)
                # Using a 30-second buffer to ensure we refresh before expiration
                self._admin_token_expiry = time.time() + token.get("expires_in", 60) - 30

                # Create admin client with the token
                self._admin_adapter = KeycloakAdmin(
                    server_url=self.configs.SERVER_URL,
                    realm_name=self.configs.REALM_NAME,
                    token=token,
                    verify=self.configs.VERIFY_SSL,
                    timeout=self.configs.TIMEOUT,
                )
                logger.debug("Admin client initialized with client credentials")
            else:
                raise UnauthenticatedError(
                    additional_data={"detail": "Neither admin credentials nor client secret provided"},
                )

        except KeycloakAuthenticationError as e:
            self._admin_adapter = None
            self._admin_token_expiry = 0
            raise UnauthenticatedError(
                additional_data={"detail": "Failed to authenticate with Keycloak service account"},
            ) from e
        except KeycloakConnectionError as e:
            self._admin_adapter = None
            self._admin_token_expiry = 0
            raise ConnectionTimeoutError("Failed to connect to Keycloak server") from e
        except KeycloakError as e:
            self._admin_adapter = None
            self._admin_token_expiry = 0
            self._handle_keycloak_exception(e, "_initialize_admin_client")

    @property
    def admin_adapter(self) -> KeycloakAdmin:
        """Get the admin adapter, refreshing it if necessary.

        Returns:
            KeycloakAdmin instance

        Raises:
            UnauthenticatedError: If admin client is not available due to authentication issues
            UnavailableError: If Keycloak service is unavailable
        """
        if not self.configs.IS_ADMIN_MODE_ENABLED or not (
            self.configs.CLIENT_SECRET_KEY or (self.configs.ADMIN_USERNAME and self.configs.ADMIN_PASSWORD)
        ):
            raise UnauthenticatedError(
                additional_data={
                    "detail": "Admin mode is disabled or neither admin credentials nor client secret provided",
                },
            )

        # Check if token is about to expire and refresh if needed
        if self._admin_adapter is None or time.time() >= self._admin_token_expiry:
            self._initialize_admin_client()

        if self._admin_adapter is None:
            raise UnavailableError("Keycloak admin client is not available")

        return self._admin_adapter

    @override
    @alru_cache(ttl=3600, maxsize=1)  # Cache for 1 hour, public key rarely changes
    async def get_public_key(self) -> PublicKeyType:
        """Get the public key used to verify tokens.

        Returns:
            JWK key object used to verify signatures

        Raises:
            ServiceUnavailableError: If Keycloak service is unavailable
            InternalError: If there's an internal error processing the public key
        """
        try:
            from jwcrypto import jwk

            keys_info = await self.openid_adapter.a_public_key()
            key = f"-----BEGIN PUBLIC KEY-----\n{keys_info}\n-----END PUBLIC KEY-----"
            return jwk.JWK.from_pem(key.encode("utf-8"))
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_public_key")
        except Exception as e:
            raise InternalError(additional_data={"operation": "get_public_key", "error": str(e)}) from e

    @override
    async def get_token(self, username: str, password: str) -> KeycloakTokenType | None:
        """Get a user token by username and password using the Resource Owner Password Credentials Grant.

        Warning:
            This method uses the direct password grant flow, which is less secure and not recommended
            for user login in production environments. Instead, prefer the web-based OAuth 2.0
            Authorization Code Flow (use `get_token_from_code`) for secure authentication.
            Use this method only for testing, administrative tasks, or specific service accounts
            where direct credential use is acceptable and properly secured.

        Args:
            username: User's username
            password: User's password

        Returns:
            Token response containing access_token, refresh_token, etc.

        Raises:
            InvalidCredentialsError: If username or password is invalid
            ServiceUnavailableError: If Keycloak service is unavailable
        """
        try:
            return await self.openid_adapter.a_token(grant_type="password", username=username, password=password)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_token")

    @override
    async def refresh_token(self, refresh_token: str) -> KeycloakTokenType | None:
        """Refresh an existing token using a refresh token.

        Args:
            refresh_token: Refresh token string

        Returns:
            New token response containing access_token, refresh_token, etc.

        Raises:
            InvalidTokenError: If refresh token is invalid or expired
            ServiceUnavailableError: If Keycloak service is unavailable
        """
        try:
            return await self.openid_adapter.a_refresh_token(refresh_token)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "refresh_token")

    @override
    async def validate_token(self, token: str) -> bool:
        """Validate if a token is still valid.

        Args:
            token: Access token to validate

        Returns:
            True if token is valid, False otherwise
        """
        # Not caching validation results as tokens are time-sensitive
        try:
            await self.openid_adapter.a_decode_token(
                token,
                key=await self.get_public_key(),
            )
        except Exception as e:
            logger.debug(f"Token validation failed: {e!s}")
            return False
        else:
            return True

    @override
    async def get_userinfo(self, token: str) -> KeycloakUserType | None:
        """Get user information from a token.

        Args:
            token: Access token

        Returns:
            User information

        Raises:
            ValueError: If getting user info fails
        """
        if not await self.validate_token(token):
            raise InvalidTokenError()
        try:
            return await self._get_userinfo_cached(token)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_userinfo")

    @alru_cache(ttl=30, maxsize=100)  # Cache for 30 seconds
    async def _get_userinfo_cached(self, token: str) -> KeycloakUserType:
        return await self.openid_adapter.a_userinfo(token)  # type: ignore[return-value]

    @override
    @alru_cache(ttl=300, maxsize=100)  # Cache for 5 minutes
    async def get_user_by_id(self, user_id: str) -> KeycloakUserType | None:
        """Get user details by user ID.

        Args:
            user_id: User's ID

        Returns:
            User details or None if not found

        Raises:
            ValueError: If getting user fails
        """
        try:
            return await self.admin_adapter.a_get_user(user_id)
        except KeycloakGetError as e:
            if e.response_code == 404:
                return None
            raise InternalError() from e
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_user_by_id")

    @override
    @alru_cache(ttl=300, maxsize=100)  # Cache for 5 minutes
    async def get_user_by_username(self, username: str) -> KeycloakUserType | None:
        """Get user details by username.

        Args:
            username: User's username

        Returns:
            User details or None if not found

        Raises:
            ValueError: If query fails
        """
        try:
            users = await self.admin_adapter.a_get_users({"username": username})
            return users[0] if users else None
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_user_by_username")

    @override
    @alru_cache(ttl=300, maxsize=100)  # Cache for 5 minutes
    async def get_user_by_email(self, email: str) -> KeycloakUserType | None:
        """Get user details by email.

        Args:
            email: User's email

        Returns:
            User details or None if not found

        Raises:
            ValueError: If query fails
        """
        try:
            users = await self.admin_adapter.a_get_users({"email": email})
            return users[0] if users else None
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_user_by_email")

    @override
    @alru_cache(ttl=300, maxsize=100)  # Cache for 5 minutes
    async def get_user_roles(self, user_id: str) -> list[KeycloakRoleType] | None:
        """Get roles assigned to a user.

        Args:
            user_id: User's ID

        Returns:
            List of roles

        Raises:
            ValueError: If getting roles fails
        """
        try:
            return await self.admin_adapter.a_get_realm_roles_of_user(user_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_user_roles")

    @override
    @alru_cache(ttl=300, maxsize=100)  # Cache for 5 minutes
    async def get_client_roles_for_user(self, user_id: str, client_id: str) -> list[KeycloakRoleType]:
        """Get client-specific roles assigned to a user.

        Args:
            user_id: User's ID
            client_id: Client ID

        Returns:
            List of client-specific roles

        Raises:
            ValueError: If getting roles fails
        """
        try:
            return await self.admin_adapter.a_get_client_roles_of_user(user_id, client_id)
        except KeycloakError as e:
            raise InternalError() from e

    @override
    async def create_user(self, user_data: dict[str, Any]) -> str | None:
        """Create a new user in Keycloak.

        Args:
            user_data: User data including username, email, etc.

        Returns:
            ID of the created user

        Raises:
            ValueError: If creating user fails
        """
        # This is a write operation, no caching needed
        try:
            user_id = await self.admin_adapter.a_create_user(user_data)

            # Clear related caches
            self.clear_all_caches()
        except KeycloakError as e:
            self._handle_user_exception(e, "create_user", user_data)
        else:
            return user_id

    @override
    async def update_user(self, user_id: str, user_data: dict[str, Any]) -> None:
        """Update user details.

        Args:
            user_id: User's ID
            user_data: User data to update

        Raises:
            ValueError: If updating user fails
        """
        # This is a write operation, no caching needed
        try:
            await self.admin_adapter.a_update_user(user_id, user_data)

            # Clear user-related caches
            self.clear_all_caches()

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "update_user")

    @override
    async def reset_password(self, user_id: str, password: str, temporary: bool = False) -> None:
        """Reset a user's password.

        Args:
            user_id: User's ID
            password: New password
            temporary: Whether the password is temporary and should be changed on next login

        Raises:
            ValueError: If password reset fails
        """
        # This is a write operation, no caching needed
        try:
            await self.admin_adapter.a_set_user_password(user_id, password, temporary)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "reset_password")

    @override
    async def assign_realm_role(self, user_id: str, role_name: str) -> None:
        """Assign a realm role to a user.

        Args:
            user_id: User's ID
            role_name: Role name to assign

        Raises:
            ValueError: If role assignment fails
        """
        # This is a write operation, no caching needed
        try:
            # Get role representation
            role = await self.admin_adapter.a_get_realm_role(role_name)
            # Assign role to user
            await self.admin_adapter.a_assign_realm_roles(user_id, [role])

            # Clear role-related caches
            if hasattr(self.get_user_roles, "cache_clear"):
                self.get_user_roles.cache_clear()

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "assign_realm_role")

    @override
    async def remove_realm_role(self, user_id: str, role_name: str) -> None:
        """Remove a realm role from a user.

        Args:
            user_id: User's ID
            role_name: Role name to remove

        Raises:
            ValueError: If role removal fails
        """
        # This is a write operation, no caching needed
        try:
            # Get role representation
            role = await self.admin_adapter.a_get_realm_role(role_name)
            # Remove role from user
            await self.admin_adapter.a_delete_realm_roles_of_user(user_id, [role])

            # Clear role-related caches
            if hasattr(self.get_user_roles, "cache_clear"):
                self.get_user_roles.cache_clear()

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "remove_realm_role")

    @override
    async def assign_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
        """Assign a client-specific role to a user.

        Args:
            user_id: User's ID
            client_id: Client ID
            role_name: Role name to assign

        Raises:
            ValueError: If role assignment fails
        """
        # This is a write operation, no caching needed
        try:
            # Get client
            client = await self.admin_adapter.a_get_client_id(client_id)
            if client is None:
                raise ValueError("client_id resolved to None")
            # Get role representation
            # Keycloak admin adapter methods accept these types at runtime
            role = await self.admin_adapter.a_get_client_role(client, role_name)
            # Assign role to user
            await self.admin_adapter.a_assign_client_role(user_id, client, [role])

            # Clear role-related caches
            if hasattr(self.get_client_roles_for_user, "cache_clear"):
                self.get_client_roles_for_user.cache_clear()

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "assign_client_role")

    @override
    async def create_realm_role(
        self,
        role_name: str,
        description: str | None = None,
        skip_exists: bool = True,
    ) -> dict[str, Any] | None:
        """Create a new realm role.

        Args:
            role_name: Role name
            description: Optional role description
            skip_exists: Skip creation if role already exists

        Returns:
            Created role details

        Raises:
            ValueError: If role creation fails
        """
        # This is a write operation, no caching needed
        try:
            role_data = {"name": role_name}
            if description:
                role_data["description"] = description

            await self.admin_adapter.a_create_realm_role(role_data, skip_exists=skip_exists)

            # Clear realm roles cache
            if hasattr(self.get_realm_roles, "cache_clear"):
                self.get_realm_roles.cache_clear()

            return await self.admin_adapter.a_get_realm_role(role_name)

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "create_realm_role")

    @override
    async def create_client_role(
        self,
        client_id: str,
        role_name: str,
        description: str | None = None,
        skip_exists: bool = True,
    ) -> dict[str, Any] | None:
        """Create a new client role.

        Args:
            client_id: Client ID or client name
            role_name: Role name
            skip_exists: Skip creation if role already exists
            description: Optional role description

        Returns:
            Created role details
        """
        # This is a write operation, no caching needed
        try:
            resolved_client_id = await self.admin_adapter.a_get_client_id(client_id)
            if resolved_client_id is None:
                raise ValueError(f"Client ID not found: {client_id}")

            # Prepare role data
            role_data = {"name": role_name}
            if description:
                role_data["description"] = description

            # Create client role
            await self.admin_adapter.a_create_client_role(resolved_client_id, role_data, skip_exists=skip_exists)

            # Clear related caches if they exist
            if hasattr(self.get_client_roles_for_user, "cache_clear"):
                self.get_client_roles_for_user.cache_clear()

            # Return created role
            return await self.admin_adapter.a_get_client_role(resolved_client_id, role_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "create_client_role")

    @override
    async def delete_realm_role(self, role_name: str) -> None:
        """Delete a realm role.

        Args:
            role_name: Role name to delete

        Raises:
            ValueError: If role deletion fails
        """
        # This is a write operation, no caching needed
        try:
            await self.admin_adapter.a_delete_realm_role(role_name)

            # Clear realm roles cache
            if hasattr(self.get_realm_roles, "cache_clear"):
                self.get_realm_roles.cache_clear()

            # We also need to clear user role caches since they might contain this role
            if hasattr(self.get_user_roles, "cache_clear"):
                self.get_user_roles.cache_clear()

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "delete_realm_role")

    @override
    @alru_cache(ttl=3600, maxsize=1)  # Cache for 1 hour
    async def get_service_account_id(self) -> str | None:
        """Get service account user ID for the current client.

        Returns:
            Service account user ID

        Raises:
            ValueError: If getting service account fails
        """
        try:
            client_id = await self.get_client_id(self.configs.CLIENT_ID)
            if client_id is None:
                return None
            service_account = await self.admin_adapter.a_get_client_service_account_user(client_id)
            return service_account.get("id")
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_service_account_id")

    @override
    @alru_cache(ttl=3600, maxsize=1)  # Cache for 1 hour
    async def get_well_known_config(self) -> dict[str, Any] | None:
        """Get the well-known OpenID configuration.

        Returns:
            OIDC configuration

        Raises:
            ValueError: If getting configuration fails
        """
        try:
            return await self.openid_adapter.a_well_known()
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_well_known_config")

    @override
    @alru_cache(ttl=3600, maxsize=1)  # Cache for 1 hour
    async def get_certs(self) -> dict[str, Any] | None:
        """Get the JWT verification certificates.

        Returns:
            Certificate information

        Raises:
            ValueError: If getting certificates fails
        """
        try:
            return await self.openid_adapter.a_certs()
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_certs")

    @override
    async def get_token_from_code(self, code: str, redirect_uri: str) -> KeycloakTokenType | None:
        """Exchange authorization code for token.

        Args:
            code: Authorization code
            redirect_uri: Redirect URI used in authorization request

        Returns:
            Token response

        Raises:
            ValueError: If token exchange fails
        """
        # Authorization codes can only be used once, don't cache
        try:
            return await self.openid_adapter.a_token(
                grant_type="authorization_code",
                code=code,
                redirect_uri=redirect_uri,
            )
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_token_from_code")

    @override
    async def get_client_credentials_token(self) -> KeycloakTokenType | None:
        """Get token using client credentials.

        Returns:
            Token response

        Raises:
            ValueError: If token acquisition fails
        """
        # Tokens are time-sensitive, don't cache
        try:
            return await self.openid_adapter.a_token(grant_type="client_credentials")
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_client_credentials_token")

    @override
    @alru_cache(ttl=30, maxsize=50)  # Cache for 30 seconds with limited entries
    async def search_users(self, query: str, max_results: int = 100) -> list[KeycloakUserType] | None:
        """Search for users by username, email, or name.

        Args:
            query: Search query
            max_results: Maximum number of results to return

        Returns:
            List of matching users

        Raises:
            ValueError: If search fails
        """
        try:
            # Try searching by different fields
            users = []

            # Search by username
            users.extend(await self.admin_adapter.a_get_users({"username": query, "max": max_results}))

            # Search by email if no results or incomplete results
            if len(users) < max_results:
                remaining = max_results - len(users)
                email_users = await self.admin_adapter.a_get_users({"email": query, "max": remaining})
                # Filter out duplicates
                user_ids = {user["id"] for user in users}
                users.extend([user for user in email_users if user["id"] not in user_ids])

            # Search by firstName if no results or incomplete results
            if len(users) < max_results:
                remaining = max_results - len(users)
                first_name_users = await self.admin_adapter.a_get_users({"firstName": query, "max": remaining})
                # Filter out duplicates
                user_ids = {user["id"] for user in users}
                users.extend([user for user in first_name_users if user["id"] not in user_ids])

            # Search by lastName if no results or incomplete results
            if len(users) < max_results:
                remaining = max_results - len(users)
                last_name_users = await self.admin_adapter.a_get_users({"lastName": query, "max": remaining})
                # Filter out duplicates
                user_ids = {user["id"] for user in users}
                users.extend([user for user in last_name_users if user["id"] not in user_ids])

            return users[:max_results]
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "search_users")

    @override
    @alru_cache(ttl=3600, maxsize=50)  # Cache for 1 hour
    async def get_client_secret(self, client_id: str) -> str | None:
        """Get client secret.

        Args:
            client_id: Client ID

        Returns:
            Client secret

        Raises:
            ValueError: If getting secret fails
        """
        try:
            client = await self.admin_adapter.a_get_client(client_id)
            return client.get("secret", "")
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_client_secret")

    @override
    @alru_cache(ttl=3600, maxsize=50)  # Cache for 1 hour
    async def get_client_id(self, client_name: str) -> str | None:
        """Get client ID by client name.

        Args:
            client_name: Name of the client

        Returns:
            Client ID

        Raises:
            ValueError: If client not found
        """
        try:
            return await self.admin_adapter.a_get_client_id(client_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_client_id")

    @override
    @alru_cache(ttl=300, maxsize=1)  # Cache for 5 minutes
    async def get_realm_roles(self) -> list[dict[str, Any]] | None:
        """Get all realm roles.

        Returns:
            List of realm roles

        Raises:
            ValueError: If getting roles fails
        """
        try:
            return await self.admin_adapter.a_get_realm_roles()
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_realm_roles")

    @override
    @alru_cache(ttl=300, maxsize=1)  # Cache for 5 minutes
    async def get_realm_role(self, role_name: str) -> dict | None:
        """Get realm role.

        Args:
            role_name: Role name
        Returns:
            A realm role

        Raises:
            ValueError: If getting role fails
        """
        try:
            return await self.admin_adapter.a_get_realm_role(role_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_realm_role")

    @override
    async def remove_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
        """Remove a client-specific role from a user.

        Args:
            user_id: User's ID
            client_id: Client ID
            role_name: Role name to remove

        Raises:
            ValueError: If role removal fails
        """
        try:
            client = await self.admin_adapter.a_get_client_id(client_id)
            if client is None:
                raise ValueError("client_id resolved to None")
            # Keycloak admin adapter methods accept these types at runtime
            role = await self.admin_adapter.a_get_client_role(client, role_name)
            await self.admin_adapter.a_delete_client_roles_of_user(user_id, client, [role])

            if hasattr(self.get_client_roles_for_user, "cache_clear"):
                self.get_client_roles_for_user.cache_clear()
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "remove_client_role")

    @override
    async def clear_user_sessions(self, user_id: str) -> None:
        """Clear all sessions for a user.

        Args:
            user_id: User's ID

        Raises:
            ValueError: If clearing sessions fails
        """
        try:
            await self.admin_adapter.a_user_logout(user_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "clear_user_sessions")

    @override
    async def logout(self, refresh_token: str) -> None:
        """Logout user by invalidating their refresh token.

        Args:
            refresh_token: Refresh token to invalidate

        Raises:
            ValueError: If logout fails
        """
        try:
            await self.openid_adapter.a_logout(refresh_token)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "logout")

    @override
    async def introspect_token(self, token: str) -> dict[str, Any] | None:
        """Introspect token to get detailed information about it.

        Args:
            token: Access token

        Returns:
            Token introspection details

        Raises:
            ValueError: If token introspection fails
        """
        try:
            return await self.openid_adapter.a_introspect(token)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "introspect_token")

    @override
    async def get_token_info(self, token: str) -> dict[str, Any] | None:
        """Decode token to get its claims.

        Args:
            token: Access token

        Returns:
            Dictionary of token claims

        Raises:
            ValueError: If token decoding fails
        """
        try:
            return await self.openid_adapter.a_decode_token(
                token,
                key=await self.get_public_key(),
            )
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_token_info")

    @override
    async def delete_user(self, user_id: str) -> None:
        """Delete a user from Keycloak by their ID.

        Args:
            user_id: The ID of the user to delete

        Raises:
            ValueError: If the deletion fails
        """
        try:
            await self.admin_adapter.a_delete_user(user_id=user_id)

            if hasattr(self.get_user_by_username, "cache_clear"):
                self.get_user_by_username.cache_clear()

            logger.info(f"Successfully deleted user with ID {user_id}")

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "delete_user")

    @override
    async def has_role(self, token: str, role_name: str) -> bool:
        """Check if a user has a specific role.

        Args:
            token: Access token
            role_name: Role name to check

        Returns:
            True if user has the role, False otherwise
        """
        # Not caching this result as token validation is time-sensitive
        try:
            user_info = await self.get_userinfo(token)
            if not user_info:
                return False

            # Check realm roles
            realm_access = user_info.get("realm_access", {})
            roles = realm_access.get("roles", [])
            if role_name in roles:
                return True

            # Check roles for the configured client
            resource_access = user_info.get("resource_access", {})
            client_roles = resource_access.get(self.configs.CLIENT_ID, {}).get("roles", [])
            if role_name in client_roles:
                return True

        except Exception as e:
            logger.debug(f"Role check failed: {e!s}")
            return False
        else:
            return False

    @override
    async def has_any_of_roles(self, token: str, role_names: frozenset[str]) -> bool:
        """Check if a user has any of the specified roles.

        Args:
            token: Access token
            role_names: Set of role names to check

        Returns:
            True if user has any of the roles, False otherwise
        """
        try:
            user_info = await self.get_userinfo(token)
            if not user_info:
                return False

            # Check realm roles first
            realm_access = user_info.get("realm_access", {})
            realm_roles = set(realm_access.get("roles", []))
            if role_names.intersection(realm_roles):
                return True

            # Check roles for the configured client
            resource_access = user_info.get("resource_access", {})
            client_roles = set(resource_access.get(self.configs.CLIENT_ID, {}).get("roles", []))
            if role_names.intersection(client_roles):
                return True

        except Exception as e:
            logger.debug(f"Role check failed: {e!s}")
            return False
        else:
            return False

    @override
    async def has_all_roles(self, token: str, role_names: frozenset[str]) -> bool:
        """Check if a user has all the specified roles.

        Args:
            token: Access token
            role_names: Set of role names to check

        Returns:
            True if user has all the roles, False otherwise
        """
        try:
            user_info = await self.get_userinfo(token)
            if not user_info:
                return False

            # Get all user roles
            all_roles = set()

            # Add realm roles
            realm_access = user_info.get("realm_access", {})
            all_roles.update(realm_access.get("roles", []))

            # Add roles from the configured client
            resource_access = user_info.get("resource_access", {})
            client_roles = resource_access.get(self.configs.CLIENT_ID, {}).get("roles", [])
            all_roles.update(client_roles)

            # Check if all required roles are present
            return role_names.issubset(all_roles)

        except Exception as e:
            logger.debug(f"All roles check failed: {e!s}")
            return False

    @override
    async def check_permissions(self, token: str, resource: str, scope: str) -> bool:
        """Check if a user has permission to access a resource with the specified scope.

        Args:
            token: Access token
            resource: Resource name
            scope: Permission scope

        Returns:
            True if permission granted, False otherwise
        """
        try:
            # Use UMA permissions endpoint to check specific resource and scope
            permissions = await self.openid_adapter.a_uma_permissions(token, permissions=f"{resource}#{scope}")

            # Check if the response indicates permission is granted
            if not permissions or not isinstance(permissions, list):
                logger.debug("No permissions returned or invalid response format")
                return False

            # Look for the specific permission in the response
            for perm in permissions:
                if perm.get("rsname") == resource and scope in perm.get("scopes", []):
                    return True

        except KeycloakError as e:
            logger.debug(f"Permission check failed with Keycloak error: {e!s}")
            return False
        except Exception as e:
            logger.debug(f"Permission check failed with unexpected error: {e!s}")
            return False
        else:
            return False

    @override
    async def create_realm(self, realm_name: str, skip_exists: bool = True, **kwargs: Any) -> dict[str, Any] | None:
        """Create a Keycloak realm with minimum required fields and optional additional config.

        Args:
            realm_name: The realm identifier (required)
            skip_exists: Skip creation if realm already exists
            kwargs: Additional optional configurations for the realm

        Returns:
            Dictionary with realm information and status

        Raises:
            InternalError: If realm creation fails
        """
        payload = {
            "realm": realm_name,
            "enabled": kwargs.get("enabled", True),
            "displayName": kwargs.get("display_name", realm_name),
        }

        # Add any additional parameters from kwargs
        for key, value in kwargs.items():
            # Skip display_name as it's already handled
            if key == "display_name":
                continue

            # Convert Python snake_case to Keycloak camelCase
            camel_key = StringUtils.snake_to_camel_case(key)
            payload[camel_key] = value

        try:
            await self.admin_adapter.a_create_realm(payload=payload, skip_exists=skip_exists)
        except KeycloakError as e:
            logger.debug(f"Failed to create realm: {e!s}")

            # Handle realm already exists with skip_exists option
            if skip_exists:
                error_message = self._extract_error_message(e).lower()
                if "already exists" in error_message and "realm" in error_message:
                    return {"realm": realm_name, "status": "already_exists", "config": payload}

            # Use the mixin to handle realm-specific errors
            self._handle_realm_exception(e, "create_realm", realm_name)
        else:
            return {"realm": realm_name, "status": "created", "config": payload}

    @override
    async def get_realm(self, realm_name: str) -> dict[str, Any] | None:
        """Get realm details by realm name.

        Args:
            realm_name: Name of the realm

        Returns:
            Realm details

        Raises:
            InternalError: If getting realm fails
        """
        try:
            return await self.admin_adapter.a_get_realm(realm_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_realm")

    @override
    async def update_realm(self, realm_name: str, **kwargs: Any) -> dict[str, Any] | None:
        """Update a realm. Kwargs are RealmRepresentation top-level attributes (e.g. displayName, organizationsEnabled).

        Args:
            realm_name: Realm name (not the realm id).
            **kwargs: RealmRepresentation attributes to update (e.g. displayName, organizationsEnabled).

        Returns:
            Response from Keycloak, or None on error (handled via exception).
        """
        try:
            return await self.admin_adapter.a_update_realm(realm_name, dict(kwargs))
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "update_realm")

    @override
    async def create_client(
        self,
        client_id: str,
        realm: str | None = None,
        skip_exists: bool = True,
        **kwargs: Any,
    ) -> dict[str, Any] | None:
        """Create a Keycloak client with minimum required fields and optional additional config.

        Args:
            client_id: The client identifier (required)
            realm: Target realm name (uses the current realm in KeycloakAdmin if not specified)
            skip_exists: Skip creation if client already exists
            kwargs: Additional optional configurations for the client

        Returns:
            Dictionary with client information

        Raises:
            InternalError: If client creation fails
        """
        original_realm = self.admin_adapter.connection.realm_name

        try:
            # Set the target realm if provided
            if realm and realm != original_realm:
                self.admin_adapter.connection.realm_name = realm

            public_client = kwargs.get("public_client", False)

            # Prepare the minimal client payload
            payload = {
                "clientId": client_id,
                "enabled": kwargs.get("enabled", True),
                "protocol": kwargs.get("protocol", "openid-connect"),
                "name": kwargs.get("name", client_id),
                "publicClient": public_client,
            }

            # Enable service accounts for confidential clients by default
            if not public_client:
                payload["serviceAccountsEnabled"] = kwargs.get("service_account_enabled", True)
                payload["clientAuthenticatorType"] = "client-secret"

            for key, value in kwargs.items():
                if key in ["enabled", "protocol", "name", "public_client", "service_account_enabled"]:
                    continue

                # Convert snake_case to camelCase
                camel_key = StringUtils.snake_to_camel_case(key)
                payload[camel_key] = value

            internal_client_id = None
            try:
                internal_client_id = await self.admin_adapter.a_create_client(payload, skip_exists=skip_exists)
            except KeycloakError as e:
                logger.debug(f"Failed to create client: {e!s}")

                # Handle client already exists with skip_exists option
                if skip_exists:
                    error_message = self._extract_error_message(e).lower()
                    if "already exists" in error_message and "client" in error_message:
                        return {
                            "client_id": client_id,
                            "status": "already_exists",
                            "realm": self.admin_adapter.connection.realm_name,
                        }

                # Use the mixin to handle client-specific errors
                client_data = {"clientId": client_id, "name": kwargs.get("name", client_id)}
                self._handle_client_exception(e, "create_client", client_data)

            return {
                "client_id": client_id,
                "internal_client_id": internal_client_id,
                "realm": self.admin_adapter.connection.realm_name,
                "status": "created",
            }

        finally:
            # Always restore the original realm
            if realm and realm != original_realm:
                self.admin_adapter.connection.realm_name = original_realm

    @override
    async def add_realm_roles_to_composite(self, composite_role_name: str, child_role_names: list[str]) -> None:
        """Add realm roles to a composite role.

        Args:
            composite_role_name: Name of the composite role
            child_role_names: List of child role names to add
        """
        try:
            child_roles = []
            for role_name in child_role_names:
                try:
                    role = await self.admin_adapter.a_get_realm_role(role_name)
                    child_roles.append(role)
                except KeycloakGetError as e:
                    if e.response_code == 404:
                        logger.warning(f"Child role not found: {role_name}")
                        continue
                    raise

            if child_roles:
                await self.admin_adapter.a_add_composite_realm_roles_to_role(
                    role_name=composite_role_name,
                    roles=child_roles,
                )
                logger.info(f"Added {len(child_roles)} realm roles to composite role: {composite_role_name}")

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "add_realm_roles_to_composite")

    @override
    async def add_client_roles_to_composite(
        self,
        composite_role_name: str,
        client_id: str,
        child_role_names: list[str],
    ) -> None:
        """Add client roles to a composite role.

        Args:
            composite_role_name: Name of the composite role
            client_id: Client ID or client name
            child_role_names: List of child role names to add
        """
        try:
            internal_client_id = await self.admin_adapter.a_get_client_id(client_id)
            if internal_client_id is None:
                raise ValueError("client_id resolved to None")

            child_roles = []
            for role_name in child_role_names:
                try:
                    # Keycloak admin adapter methods accept these types at runtime
                    role = await self.admin_adapter.a_get_client_role(internal_client_id, role_name)
                    child_roles.append(role)
                except KeycloakGetError as e:
                    if e.response_code == 404:
                        logger.warning(f"Client role not found: {role_name}")
                        continue
                    raise

            if child_roles:
                if internal_client_id is None:
                    raise ValueError("Client ID not found")
                resolved_client_id: str = internal_client_id
                await self.admin_adapter.a_add_composite_client_roles_to_role(
                    role_name=composite_role_name,
                    client_role_id=resolved_client_id,
                    roles=child_roles,
                )
                logger.info(f"Added {len(child_roles)} client roles to composite role: {composite_role_name}")

        except KeycloakError as e:
            self._handle_keycloak_exception(e, "add_client_roles_to_composite")

    @override
    async def get_composite_realm_roles(self, role_name: str) -> list[dict[str, Any]] | None:
        """Get composite roles for a realm role.

        Args:
            role_name: Name of the role

        Returns:
            List of composite roles
        """
        try:
            return await self.admin_adapter.a_get_composite_realm_roles_of_role(role_name)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_composite_realm_roles")

    @override
    async def get_organizations(self, query: dict | None = None) -> list[dict[str, Any]]:
        """Fetch all organizations, optionally filtered by query parameters."""
        try:
            return await self.admin_adapter.a_get_organizations(query=query)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_organizations")

    @override
    async def get_organization(self, organization_id: str) -> dict[str, Any]:
        """Get representation of the organization by ID."""
        try:
            return await self.admin_adapter.a_get_organization(organization_id=organization_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_organization")

    @override
    async def create_organization(self, name: str, alias: str, **kwargs: Any) -> str | None:
        """Create a new organization. Name and alias must be unique. Returns org_id."""
        try:
            payload = {"name": name, "alias": alias}
            for key, value in kwargs.items():
                if key in ["name", "alias"]:
                    continue

                # Convert snake_case to camelCase
                camel_key = StringUtils.snake_to_camel_case(key)
                payload[camel_key] = value

            return await self.admin_adapter.a_create_organization(payload=payload)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "create_organization")

    @override
    async def update_organization(self, organization_id: str, **kwargs: Any) -> dict[str, Any]:
        """Update an existing organization. Kwargs are organization attributes (e.g. name, alias)."""
        try:
            payload = {}
            for key, value in kwargs.items():
                # Convert snake_case to camelCase
                camel_key = StringUtils.snake_to_camel_case(key)
                payload[camel_key] = value

            return await self.admin_adapter.a_update_organization(organization_id=organization_id, payload=payload)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "update_organization")

    @override
    async def delete_organization(self, organization_id: str) -> dict[str, Any]:
        """Delete an organization."""
        try:
            return await self.admin_adapter.a_delete_organization(organization_id=organization_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "delete_organization")

    @override
    async def get_user_organizations(self, user_id: str) -> list[dict[str, Any]]:
        """Get organizations by user id."""
        try:
            return await self.admin_adapter.a_get_user_organizations(user_id=user_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_user_organizations")

    @override
    async def get_organization_members(self, organization_id: str, query: dict | None = None) -> list[dict[str, Any]]:
        """Get members by organization id, optionally filtered by query parameters."""
        try:
            return await self.admin_adapter.a_get_organization_members(organization_id=organization_id, query=query)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_organization_members")

    @override
    async def get_organization_members_count(self, organization_id: str) -> int:
        """Get the number of members in the organization."""
        try:
            return await self.admin_adapter.a_get_organization_members_count(organization_id=organization_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "get_organization_members_count")

    @override
    async def organization_user_add(self, user_id: str, organization_id: str) -> bytes:
        """Add a user to an organization."""
        try:
            return await self.admin_adapter.a_organization_user_add(user_id=user_id, organization_id=organization_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "organization_user_add")

    @override
    async def organization_user_remove(self, user_id: str, organization_id: str) -> dict[str, Any]:
        """Remove a user from an organization."""
        try:
            return await self.admin_adapter.a_organization_user_remove(user_id=user_id, organization_id=organization_id)
        except KeycloakError as e:
            self._handle_keycloak_exception(e, "organization_user_remove")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.configs instance-attribute

configs: KeycloakConfig = (
    KEYCLOAK
    if keycloak_configs is None
    else keycloak_configs
)

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.openid_adapter instance-attribute

openid_adapter = _get_openid_client(configs)

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.admin_adapter property

admin_adapter: KeycloakAdmin

Get the admin adapter, refreshing it if necessary.

Returns:

Type	Description
KeycloakAdmin	KeycloakAdmin instance

Raises:

Type	Description
UnauthenticatedError	If admin client is not available due to authentication issues
UnavailableError	If Keycloak service is unavailable

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.clear_all_caches

clear_all_caches() -> None

Clear all cached values.

Source code in archipy/adapters/keycloak/adapters.py

def clear_all_caches(self) -> None:
    """Clear all cached values."""
    for attr_name in dir(self):
        attr = getattr(self, attr_name)
        if hasattr(attr, "cache_clear"):
            attr.cache_clear()

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_public_key async

get_public_key() -> PublicKeyType

Get the public key used to verify tokens.

Returns:

Type	Description
PublicKeyType	JWK key object used to verify signatures

Raises:

Type	Description
ServiceUnavailableError	If Keycloak service is unavailable
InternalError	If there's an internal error processing the public key

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=3600, maxsize=1)  # Cache for 1 hour, public key rarely changes
async def get_public_key(self) -> PublicKeyType:
    """Get the public key used to verify tokens.

    Returns:
        JWK key object used to verify signatures

    Raises:
        ServiceUnavailableError: If Keycloak service is unavailable
        InternalError: If there's an internal error processing the public key
    """
    try:
        from jwcrypto import jwk

        keys_info = await self.openid_adapter.a_public_key()
        key = f"-----BEGIN PUBLIC KEY-----\n{keys_info}\n-----END PUBLIC KEY-----"
        return jwk.JWK.from_pem(key.encode("utf-8"))
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_public_key")
    except Exception as e:
        raise InternalError(additional_data={"operation": "get_public_key", "error": str(e)}) from e

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_token async

get_token(
    username: str, password: str
) -> KeycloakTokenType | None

Get a user token by username and password using the Resource Owner Password Credentials Grant.

Warning

This method uses the direct password grant flow, which is less secure and not recommended for user login in production environments. Instead, prefer the web-based OAuth 2.0 Authorization Code Flow (use get_token_from_code) for secure authentication. Use this method only for testing, administrative tasks, or specific service accounts where direct credential use is acceptable and properly secured.

Parameters:

Name	Type	Description	Default
username	str	User's username	required
password	str	User's password	required

Returns:

Type	Description
KeycloakTokenType | None	Token response containing access_token, refresh_token, etc.

Raises:

Type	Description
InvalidCredentialsError	If username or password is invalid
ServiceUnavailableError	If Keycloak service is unavailable

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_token(self, username: str, password: str) -> KeycloakTokenType | None:
    """Get a user token by username and password using the Resource Owner Password Credentials Grant.

    Warning:
        This method uses the direct password grant flow, which is less secure and not recommended
        for user login in production environments. Instead, prefer the web-based OAuth 2.0
        Authorization Code Flow (use `get_token_from_code`) for secure authentication.
        Use this method only for testing, administrative tasks, or specific service accounts
        where direct credential use is acceptable and properly secured.

    Args:
        username: User's username
        password: User's password

    Returns:
        Token response containing access_token, refresh_token, etc.

    Raises:
        InvalidCredentialsError: If username or password is invalid
        ServiceUnavailableError: If Keycloak service is unavailable
    """
    try:
        return await self.openid_adapter.a_token(grant_type="password", username=username, password=password)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_token")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.refresh_token async

refresh_token(
    refresh_token: str,
) -> KeycloakTokenType | None

Refresh an existing token using a refresh token.

Parameters:

Name	Type	Description	Default
refresh_token	str	Refresh token string	required

Returns:

Type	Description
KeycloakTokenType | None	New token response containing access_token, refresh_token, etc.

Raises:

Type	Description
InvalidTokenError	If refresh token is invalid or expired
ServiceUnavailableError	If Keycloak service is unavailable

Source code in archipy/adapters/keycloak/adapters.py

@override
async def refresh_token(self, refresh_token: str) -> KeycloakTokenType | None:
    """Refresh an existing token using a refresh token.

    Args:
        refresh_token: Refresh token string

    Returns:
        New token response containing access_token, refresh_token, etc.

    Raises:
        InvalidTokenError: If refresh token is invalid or expired
        ServiceUnavailableError: If Keycloak service is unavailable
    """
    try:
        return await self.openid_adapter.a_refresh_token(refresh_token)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "refresh_token")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.validate_token async

validate_token(token: str) -> bool

Validate if a token is still valid.

Parameters:

Name	Type	Description	Default
token	str	Access token to validate	required

Returns:

Type	Description
bool	True if token is valid, False otherwise

Source code in archipy/adapters/keycloak/adapters.py

@override
async def validate_token(self, token: str) -> bool:
    """Validate if a token is still valid.

    Args:
        token: Access token to validate

    Returns:
        True if token is valid, False otherwise
    """
    # Not caching validation results as tokens are time-sensitive
    try:
        await self.openid_adapter.a_decode_token(
            token,
            key=await self.get_public_key(),
        )
    except Exception as e:
        logger.debug(f"Token validation failed: {e!s}")
        return False
    else:
        return True

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_userinfo async

get_userinfo(token: str) -> KeycloakUserType | None

Get user information from a token.

Parameters:

Name	Type	Description	Default
token	str	Access token	required

Returns:

Type	Description
KeycloakUserType | None	User information

Raises:

Type	Description
ValueError	If getting user info fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_userinfo(self, token: str) -> KeycloakUserType | None:
    """Get user information from a token.

    Args:
        token: Access token

    Returns:
        User information

    Raises:
        ValueError: If getting user info fails
    """
    if not await self.validate_token(token):
        raise InvalidTokenError()
    try:
        return await self._get_userinfo_cached(token)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_userinfo")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_user_by_id async

get_user_by_id(user_id: str) -> KeycloakUserType | None

Get user details by user ID.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required

Returns:

Type	Description
KeycloakUserType | None	User details or None if not found

Raises:

Type	Description
ValueError	If getting user fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=300, maxsize=100)  # Cache for 5 minutes
async def get_user_by_id(self, user_id: str) -> KeycloakUserType | None:
    """Get user details by user ID.

    Args:
        user_id: User's ID

    Returns:
        User details or None if not found

    Raises:
        ValueError: If getting user fails
    """
    try:
        return await self.admin_adapter.a_get_user(user_id)
    except KeycloakGetError as e:
        if e.response_code == 404:
            return None
        raise InternalError() from e
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_user_by_id")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_user_by_username async

get_user_by_username(
    username: str,
) -> KeycloakUserType | None

Get user details by username.

Parameters:

Name	Type	Description	Default
username	str	User's username	required

Returns:

Type	Description
KeycloakUserType | None	User details or None if not found

Raises:

Type	Description
ValueError	If query fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=300, maxsize=100)  # Cache for 5 minutes
async def get_user_by_username(self, username: str) -> KeycloakUserType | None:
    """Get user details by username.

    Args:
        username: User's username

    Returns:
        User details or None if not found

    Raises:
        ValueError: If query fails
    """
    try:
        users = await self.admin_adapter.a_get_users({"username": username})
        return users[0] if users else None
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_user_by_username")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_user_by_email async

get_user_by_email(email: str) -> KeycloakUserType | None

Get user details by email.

Parameters:

Name	Type	Description	Default
email	str	User's email	required

Returns:

Type	Description
KeycloakUserType | None	User details or None if not found

Raises:

Type	Description
ValueError	If query fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=300, maxsize=100)  # Cache for 5 minutes
async def get_user_by_email(self, email: str) -> KeycloakUserType | None:
    """Get user details by email.

    Args:
        email: User's email

    Returns:
        User details or None if not found

    Raises:
        ValueError: If query fails
    """
    try:
        users = await self.admin_adapter.a_get_users({"email": email})
        return users[0] if users else None
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_user_by_email")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_user_roles async

get_user_roles(
    user_id: str,
) -> list[KeycloakRoleType] | None

Get roles assigned to a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required

Returns:

Type	Description
list[KeycloakRoleType] | None	List of roles

Raises:

Type	Description
ValueError	If getting roles fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=300, maxsize=100)  # Cache for 5 minutes
async def get_user_roles(self, user_id: str) -> list[KeycloakRoleType] | None:
    """Get roles assigned to a user.

    Args:
        user_id: User's ID

    Returns:
        List of roles

    Raises:
        ValueError: If getting roles fails
    """
    try:
        return await self.admin_adapter.a_get_realm_roles_of_user(user_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_user_roles")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_client_roles_for_user async

get_client_roles_for_user(
    user_id: str, client_id: str
) -> list[KeycloakRoleType]

Get client-specific roles assigned to a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
client_id	str	Client ID	required

Returns:

Type	Description
list[KeycloakRoleType]	List of client-specific roles

Raises:

Type	Description
ValueError	If getting roles fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=300, maxsize=100)  # Cache for 5 minutes
async def get_client_roles_for_user(self, user_id: str, client_id: str) -> list[KeycloakRoleType]:
    """Get client-specific roles assigned to a user.

    Args:
        user_id: User's ID
        client_id: Client ID

    Returns:
        List of client-specific roles

    Raises:
        ValueError: If getting roles fails
    """
    try:
        return await self.admin_adapter.a_get_client_roles_of_user(user_id, client_id)
    except KeycloakError as e:
        raise InternalError() from e

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.create_user async

create_user(user_data: dict[str, Any]) -> str | None

Create a new user in Keycloak.

Parameters:

Name	Type	Description	Default
user_data	dict[str, Any]	User data including username, email, etc.	required

Returns:

Type	Description
str | None	ID of the created user

Raises:

Type	Description
ValueError	If creating user fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def create_user(self, user_data: dict[str, Any]) -> str | None:
    """Create a new user in Keycloak.

    Args:
        user_data: User data including username, email, etc.

    Returns:
        ID of the created user

    Raises:
        ValueError: If creating user fails
    """
    # This is a write operation, no caching needed
    try:
        user_id = await self.admin_adapter.a_create_user(user_data)

        # Clear related caches
        self.clear_all_caches()
    except KeycloakError as e:
        self._handle_user_exception(e, "create_user", user_data)
    else:
        return user_id

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.update_user async

update_user(
    user_id: str, user_data: dict[str, Any]
) -> None

Update user details.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
user_data	dict[str, Any]	User data to update	required

Raises:

Type	Description
ValueError	If updating user fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def update_user(self, user_id: str, user_data: dict[str, Any]) -> None:
    """Update user details.

    Args:
        user_id: User's ID
        user_data: User data to update

    Raises:
        ValueError: If updating user fails
    """
    # This is a write operation, no caching needed
    try:
        await self.admin_adapter.a_update_user(user_id, user_data)

        # Clear user-related caches
        self.clear_all_caches()

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "update_user")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.reset_password async

reset_password(
    user_id: str, password: str, temporary: bool = False
) -> None

Reset a user's password.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
password	str	New password	required
temporary	bool	Whether the password is temporary and should be changed on next login	False

Raises:

Type	Description
ValueError	If password reset fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def reset_password(self, user_id: str, password: str, temporary: bool = False) -> None:
    """Reset a user's password.

    Args:
        user_id: User's ID
        password: New password
        temporary: Whether the password is temporary and should be changed on next login

    Raises:
        ValueError: If password reset fails
    """
    # This is a write operation, no caching needed
    try:
        await self.admin_adapter.a_set_user_password(user_id, password, temporary)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "reset_password")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.assign_realm_role async

assign_realm_role(user_id: str, role_name: str) -> None

Assign a realm role to a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
role_name	str	Role name to assign	required

Raises:

Type	Description
ValueError	If role assignment fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def assign_realm_role(self, user_id: str, role_name: str) -> None:
    """Assign a realm role to a user.

    Args:
        user_id: User's ID
        role_name: Role name to assign

    Raises:
        ValueError: If role assignment fails
    """
    # This is a write operation, no caching needed
    try:
        # Get role representation
        role = await self.admin_adapter.a_get_realm_role(role_name)
        # Assign role to user
        await self.admin_adapter.a_assign_realm_roles(user_id, [role])

        # Clear role-related caches
        if hasattr(self.get_user_roles, "cache_clear"):
            self.get_user_roles.cache_clear()

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "assign_realm_role")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.remove_realm_role async

remove_realm_role(user_id: str, role_name: str) -> None

Remove a realm role from a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
role_name	str	Role name to remove	required

Raises:

Type	Description
ValueError	If role removal fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def remove_realm_role(self, user_id: str, role_name: str) -> None:
    """Remove a realm role from a user.

    Args:
        user_id: User's ID
        role_name: Role name to remove

    Raises:
        ValueError: If role removal fails
    """
    # This is a write operation, no caching needed
    try:
        # Get role representation
        role = await self.admin_adapter.a_get_realm_role(role_name)
        # Remove role from user
        await self.admin_adapter.a_delete_realm_roles_of_user(user_id, [role])

        # Clear role-related caches
        if hasattr(self.get_user_roles, "cache_clear"):
            self.get_user_roles.cache_clear()

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "remove_realm_role")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.assign_client_role async

assign_client_role(
    user_id: str, client_id: str, role_name: str
) -> None

Assign a client-specific role to a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
client_id	str	Client ID	required
role_name	str	Role name to assign	required

Raises:

Type	Description
ValueError	If role assignment fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def assign_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
    """Assign a client-specific role to a user.

    Args:
        user_id: User's ID
        client_id: Client ID
        role_name: Role name to assign

    Raises:
        ValueError: If role assignment fails
    """
    # This is a write operation, no caching needed
    try:
        # Get client
        client = await self.admin_adapter.a_get_client_id(client_id)
        if client is None:
            raise ValueError("client_id resolved to None")
        # Get role representation
        # Keycloak admin adapter methods accept these types at runtime
        role = await self.admin_adapter.a_get_client_role(client, role_name)
        # Assign role to user
        await self.admin_adapter.a_assign_client_role(user_id, client, [role])

        # Clear role-related caches
        if hasattr(self.get_client_roles_for_user, "cache_clear"):
            self.get_client_roles_for_user.cache_clear()

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "assign_client_role")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.create_realm_role async

create_realm_role(
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None

Create a new realm role.

Parameters:

Name	Type	Description	Default
role_name	str	Role name	required
description	str | None	Optional role description	None
skip_exists	bool	Skip creation if role already exists	True

Returns:

Type	Description
dict[str, Any] | None	Created role details

Raises:

Type	Description
ValueError	If role creation fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def create_realm_role(
    self,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None:
    """Create a new realm role.

    Args:
        role_name: Role name
        description: Optional role description
        skip_exists: Skip creation if role already exists

    Returns:
        Created role details

    Raises:
        ValueError: If role creation fails
    """
    # This is a write operation, no caching needed
    try:
        role_data = {"name": role_name}
        if description:
            role_data["description"] = description

        await self.admin_adapter.a_create_realm_role(role_data, skip_exists=skip_exists)

        # Clear realm roles cache
        if hasattr(self.get_realm_roles, "cache_clear"):
            self.get_realm_roles.cache_clear()

        return await self.admin_adapter.a_get_realm_role(role_name)

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "create_realm_role")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.create_client_role async

create_client_role(
    client_id: str,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None

Create a new client role.

Parameters:

Name	Type	Description	Default
client_id	str	Client ID or client name	required
role_name	str	Role name	required
skip_exists	bool	Skip creation if role already exists	True
description	str | None	Optional role description	None

Returns:

Type	Description
dict[str, Any] | None	Created role details

Source code in archipy/adapters/keycloak/adapters.py

@override
async def create_client_role(
    self,
    client_id: str,
    role_name: str,
    description: str | None = None,
    skip_exists: bool = True,
) -> dict[str, Any] | None:
    """Create a new client role.

    Args:
        client_id: Client ID or client name
        role_name: Role name
        skip_exists: Skip creation if role already exists
        description: Optional role description

    Returns:
        Created role details
    """
    # This is a write operation, no caching needed
    try:
        resolved_client_id = await self.admin_adapter.a_get_client_id(client_id)
        if resolved_client_id is None:
            raise ValueError(f"Client ID not found: {client_id}")

        # Prepare role data
        role_data = {"name": role_name}
        if description:
            role_data["description"] = description

        # Create client role
        await self.admin_adapter.a_create_client_role(resolved_client_id, role_data, skip_exists=skip_exists)

        # Clear related caches if they exist
        if hasattr(self.get_client_roles_for_user, "cache_clear"):
            self.get_client_roles_for_user.cache_clear()

        # Return created role
        return await self.admin_adapter.a_get_client_role(resolved_client_id, role_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "create_client_role")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.delete_realm_role async

delete_realm_role(role_name: str) -> None

Delete a realm role.

Parameters:

Name	Type	Description	Default
role_name	str	Role name to delete	required

Raises:

Type	Description
ValueError	If role deletion fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def delete_realm_role(self, role_name: str) -> None:
    """Delete a realm role.

    Args:
        role_name: Role name to delete

    Raises:
        ValueError: If role deletion fails
    """
    # This is a write operation, no caching needed
    try:
        await self.admin_adapter.a_delete_realm_role(role_name)

        # Clear realm roles cache
        if hasattr(self.get_realm_roles, "cache_clear"):
            self.get_realm_roles.cache_clear()

        # We also need to clear user role caches since they might contain this role
        if hasattr(self.get_user_roles, "cache_clear"):
            self.get_user_roles.cache_clear()

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "delete_realm_role")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_service_account_id async

get_service_account_id() -> str | None

Get service account user ID for the current client.

Returns:

Type	Description
str | None	Service account user ID

Raises:

Type	Description
ValueError	If getting service account fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=3600, maxsize=1)  # Cache for 1 hour
async def get_service_account_id(self) -> str | None:
    """Get service account user ID for the current client.

    Returns:
        Service account user ID

    Raises:
        ValueError: If getting service account fails
    """
    try:
        client_id = await self.get_client_id(self.configs.CLIENT_ID)
        if client_id is None:
            return None
        service_account = await self.admin_adapter.a_get_client_service_account_user(client_id)
        return service_account.get("id")
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_service_account_id")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_well_known_config async

get_well_known_config() -> dict[str, Any] | None

Get the well-known OpenID configuration.

Returns:

Type	Description
dict[str, Any] | None	OIDC configuration

Raises:

Type	Description
ValueError	If getting configuration fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=3600, maxsize=1)  # Cache for 1 hour
async def get_well_known_config(self) -> dict[str, Any] | None:
    """Get the well-known OpenID configuration.

    Returns:
        OIDC configuration

    Raises:
        ValueError: If getting configuration fails
    """
    try:
        return await self.openid_adapter.a_well_known()
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_well_known_config")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_certs async

get_certs() -> dict[str, Any] | None

Get the JWT verification certificates.

Returns:

Type	Description
dict[str, Any] | None	Certificate information

Raises:

Type	Description
ValueError	If getting certificates fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=3600, maxsize=1)  # Cache for 1 hour
async def get_certs(self) -> dict[str, Any] | None:
    """Get the JWT verification certificates.

    Returns:
        Certificate information

    Raises:
        ValueError: If getting certificates fails
    """
    try:
        return await self.openid_adapter.a_certs()
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_certs")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_token_from_code async

get_token_from_code(
    code: str, redirect_uri: str
) -> KeycloakTokenType | None

Exchange authorization code for token.

Parameters:

Name	Type	Description	Default
code	str	Authorization code	required
redirect_uri	str	Redirect URI used in authorization request	required

Returns:

Type	Description
KeycloakTokenType | None	Token response

Raises:

Type	Description
ValueError	If token exchange fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_token_from_code(self, code: str, redirect_uri: str) -> KeycloakTokenType | None:
    """Exchange authorization code for token.

    Args:
        code: Authorization code
        redirect_uri: Redirect URI used in authorization request

    Returns:
        Token response

    Raises:
        ValueError: If token exchange fails
    """
    # Authorization codes can only be used once, don't cache
    try:
        return await self.openid_adapter.a_token(
            grant_type="authorization_code",
            code=code,
            redirect_uri=redirect_uri,
        )
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_token_from_code")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_client_credentials_token async

get_client_credentials_token() -> KeycloakTokenType | None

Get token using client credentials.

Returns:

Type	Description
KeycloakTokenType | None	Token response

Raises:

Type	Description
ValueError	If token acquisition fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_client_credentials_token(self) -> KeycloakTokenType | None:
    """Get token using client credentials.

    Returns:
        Token response

    Raises:
        ValueError: If token acquisition fails
    """
    # Tokens are time-sensitive, don't cache
    try:
        return await self.openid_adapter.a_token(grant_type="client_credentials")
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_client_credentials_token")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.search_users async

search_users(
    query: str, max_results: int = 100
) -> list[KeycloakUserType] | None

Search for users by username, email, or name.

Parameters:

Name	Type	Description	Default
query	str	Search query	required
max_results	int	Maximum number of results to return	100

Returns:

Type	Description
list[KeycloakUserType] | None	List of matching users

Raises:

Type	Description
ValueError	If search fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=30, maxsize=50)  # Cache for 30 seconds with limited entries
async def search_users(self, query: str, max_results: int = 100) -> list[KeycloakUserType] | None:
    """Search for users by username, email, or name.

    Args:
        query: Search query
        max_results: Maximum number of results to return

    Returns:
        List of matching users

    Raises:
        ValueError: If search fails
    """
    try:
        # Try searching by different fields
        users = []

        # Search by username
        users.extend(await self.admin_adapter.a_get_users({"username": query, "max": max_results}))

        # Search by email if no results or incomplete results
        if len(users) < max_results:
            remaining = max_results - len(users)
            email_users = await self.admin_adapter.a_get_users({"email": query, "max": remaining})
            # Filter out duplicates
            user_ids = {user["id"] for user in users}
            users.extend([user for user in email_users if user["id"] not in user_ids])

        # Search by firstName if no results or incomplete results
        if len(users) < max_results:
            remaining = max_results - len(users)
            first_name_users = await self.admin_adapter.a_get_users({"firstName": query, "max": remaining})
            # Filter out duplicates
            user_ids = {user["id"] for user in users}
            users.extend([user for user in first_name_users if user["id"] not in user_ids])

        # Search by lastName if no results or incomplete results
        if len(users) < max_results:
            remaining = max_results - len(users)
            last_name_users = await self.admin_adapter.a_get_users({"lastName": query, "max": remaining})
            # Filter out duplicates
            user_ids = {user["id"] for user in users}
            users.extend([user for user in last_name_users if user["id"] not in user_ids])

        return users[:max_results]
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "search_users")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_client_secret async

get_client_secret(client_id: str) -> str | None

Get client secret.

Parameters:

Name	Type	Description	Default
client_id	str	Client ID	required

Returns:

Type	Description
str | None	Client secret

Raises:

Type	Description
ValueError	If getting secret fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=3600, maxsize=50)  # Cache for 1 hour
async def get_client_secret(self, client_id: str) -> str | None:
    """Get client secret.

    Args:
        client_id: Client ID

    Returns:
        Client secret

    Raises:
        ValueError: If getting secret fails
    """
    try:
        client = await self.admin_adapter.a_get_client(client_id)
        return client.get("secret", "")
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_client_secret")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_client_id async

get_client_id(client_name: str) -> str | None

Get client ID by client name.

Parameters:

Name	Type	Description	Default
client_name	str	Name of the client	required

Returns:

Type	Description
str | None	Client ID

Raises:

Type	Description
ValueError	If client not found

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=3600, maxsize=50)  # Cache for 1 hour
async def get_client_id(self, client_name: str) -> str | None:
    """Get client ID by client name.

    Args:
        client_name: Name of the client

    Returns:
        Client ID

    Raises:
        ValueError: If client not found
    """
    try:
        return await self.admin_adapter.a_get_client_id(client_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_client_id")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_realm_roles async

get_realm_roles() -> list[dict[str, Any]] | None

Get all realm roles.

Returns:

Type	Description
list[dict[str, Any]] | None	List of realm roles

Raises:

Type	Description
ValueError	If getting roles fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=300, maxsize=1)  # Cache for 5 minutes
async def get_realm_roles(self) -> list[dict[str, Any]] | None:
    """Get all realm roles.

    Returns:
        List of realm roles

    Raises:
        ValueError: If getting roles fails
    """
    try:
        return await self.admin_adapter.a_get_realm_roles()
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_realm_roles")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_realm_role async

get_realm_role(role_name: str) -> dict | None

Get realm role.

Parameters:

Name	Type	Description	Default
role_name	str	Role name	required

Returns: A realm role

Raises:

Type	Description
ValueError	If getting role fails

Source code in archipy/adapters/keycloak/adapters.py

@override
@alru_cache(ttl=300, maxsize=1)  # Cache for 5 minutes
async def get_realm_role(self, role_name: str) -> dict | None:
    """Get realm role.

    Args:
        role_name: Role name
    Returns:
        A realm role

    Raises:
        ValueError: If getting role fails
    """
    try:
        return await self.admin_adapter.a_get_realm_role(role_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_realm_role")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.remove_client_role async

remove_client_role(
    user_id: str, client_id: str, role_name: str
) -> None

Remove a client-specific role from a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required
client_id	str	Client ID	required
role_name	str	Role name to remove	required

Raises:

Type	Description
ValueError	If role removal fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def remove_client_role(self, user_id: str, client_id: str, role_name: str) -> None:
    """Remove a client-specific role from a user.

    Args:
        user_id: User's ID
        client_id: Client ID
        role_name: Role name to remove

    Raises:
        ValueError: If role removal fails
    """
    try:
        client = await self.admin_adapter.a_get_client_id(client_id)
        if client is None:
            raise ValueError("client_id resolved to None")
        # Keycloak admin adapter methods accept these types at runtime
        role = await self.admin_adapter.a_get_client_role(client, role_name)
        await self.admin_adapter.a_delete_client_roles_of_user(user_id, client, [role])

        if hasattr(self.get_client_roles_for_user, "cache_clear"):
            self.get_client_roles_for_user.cache_clear()
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "remove_client_role")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.clear_user_sessions async

clear_user_sessions(user_id: str) -> None

Clear all sessions for a user.

Parameters:

Name	Type	Description	Default
user_id	str	User's ID	required

Raises:

Type	Description
ValueError	If clearing sessions fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def clear_user_sessions(self, user_id: str) -> None:
    """Clear all sessions for a user.

    Args:
        user_id: User's ID

    Raises:
        ValueError: If clearing sessions fails
    """
    try:
        await self.admin_adapter.a_user_logout(user_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "clear_user_sessions")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.logout async

logout(refresh_token: str) -> None

Logout user by invalidating their refresh token.

Parameters:

Name	Type	Description	Default
refresh_token	str	Refresh token to invalidate	required

Raises:

Type	Description
ValueError	If logout fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def logout(self, refresh_token: str) -> None:
    """Logout user by invalidating their refresh token.

    Args:
        refresh_token: Refresh token to invalidate

    Raises:
        ValueError: If logout fails
    """
    try:
        await self.openid_adapter.a_logout(refresh_token)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "logout")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.introspect_token async

introspect_token(token: str) -> dict[str, Any] | None

Introspect token to get detailed information about it.

Parameters:

Name	Type	Description	Default
token	str	Access token	required

Returns:

Type	Description
dict[str, Any] | None	Token introspection details

Raises:

Type	Description
ValueError	If token introspection fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def introspect_token(self, token: str) -> dict[str, Any] | None:
    """Introspect token to get detailed information about it.

    Args:
        token: Access token

    Returns:
        Token introspection details

    Raises:
        ValueError: If token introspection fails
    """
    try:
        return await self.openid_adapter.a_introspect(token)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "introspect_token")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_token_info async

get_token_info(token: str) -> dict[str, Any] | None

Decode token to get its claims.

Parameters:

Name	Type	Description	Default
token	str	Access token	required

Returns:

Type	Description
dict[str, Any] | None	Dictionary of token claims

Raises:

Type	Description
ValueError	If token decoding fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_token_info(self, token: str) -> dict[str, Any] | None:
    """Decode token to get its claims.

    Args:
        token: Access token

    Returns:
        Dictionary of token claims

    Raises:
        ValueError: If token decoding fails
    """
    try:
        return await self.openid_adapter.a_decode_token(
            token,
            key=await self.get_public_key(),
        )
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_token_info")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.delete_user async

delete_user(user_id: str) -> None

Delete a user from Keycloak by their ID.

Parameters:

Name	Type	Description	Default
user_id	str	The ID of the user to delete	required

Raises:

Type	Description
ValueError	If the deletion fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def delete_user(self, user_id: str) -> None:
    """Delete a user from Keycloak by their ID.

    Args:
        user_id: The ID of the user to delete

    Raises:
        ValueError: If the deletion fails
    """
    try:
        await self.admin_adapter.a_delete_user(user_id=user_id)

        if hasattr(self.get_user_by_username, "cache_clear"):
            self.get_user_by_username.cache_clear()

        logger.info(f"Successfully deleted user with ID {user_id}")

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "delete_user")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.has_role async

has_role(token: str, role_name: str) -> bool

Check if a user has a specific role.

Parameters:

Name	Type	Description	Default
token	str	Access token	required
role_name	str	Role name to check	required

Returns:

Type	Description
bool	True if user has the role, False otherwise

Source code in archipy/adapters/keycloak/adapters.py

@override
async def has_role(self, token: str, role_name: str) -> bool:
    """Check if a user has a specific role.

    Args:
        token: Access token
        role_name: Role name to check

    Returns:
        True if user has the role, False otherwise
    """
    # Not caching this result as token validation is time-sensitive
    try:
        user_info = await self.get_userinfo(token)
        if not user_info:
            return False

        # Check realm roles
        realm_access = user_info.get("realm_access", {})
        roles = realm_access.get("roles", [])
        if role_name in roles:
            return True

        # Check roles for the configured client
        resource_access = user_info.get("resource_access", {})
        client_roles = resource_access.get(self.configs.CLIENT_ID, {}).get("roles", [])
        if role_name in client_roles:
            return True

    except Exception as e:
        logger.debug(f"Role check failed: {e!s}")
        return False
    else:
        return False

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.has_any_of_roles async

has_any_of_roles(
    token: str, role_names: frozenset[str]
) -> bool

Check if a user has any of the specified roles.

Parameters:

Name	Type	Description	Default
token	str	Access token	required
role_names	frozenset[str]	Set of role names to check	required

Returns:

Type	Description
bool	True if user has any of the roles, False otherwise

Source code in archipy/adapters/keycloak/adapters.py

@override
async def has_any_of_roles(self, token: str, role_names: frozenset[str]) -> bool:
    """Check if a user has any of the specified roles.

    Args:
        token: Access token
        role_names: Set of role names to check

    Returns:
        True if user has any of the roles, False otherwise
    """
    try:
        user_info = await self.get_userinfo(token)
        if not user_info:
            return False

        # Check realm roles first
        realm_access = user_info.get("realm_access", {})
        realm_roles = set(realm_access.get("roles", []))
        if role_names.intersection(realm_roles):
            return True

        # Check roles for the configured client
        resource_access = user_info.get("resource_access", {})
        client_roles = set(resource_access.get(self.configs.CLIENT_ID, {}).get("roles", []))
        if role_names.intersection(client_roles):
            return True

    except Exception as e:
        logger.debug(f"Role check failed: {e!s}")
        return False
    else:
        return False

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.has_all_roles async

has_all_roles(
    token: str, role_names: frozenset[str]
) -> bool

Check if a user has all the specified roles.

Parameters:

Name	Type	Description	Default
token	str	Access token	required
role_names	frozenset[str]	Set of role names to check	required

Returns:

Type	Description
bool	True if user has all the roles, False otherwise

Source code in archipy/adapters/keycloak/adapters.py

@override
async def has_all_roles(self, token: str, role_names: frozenset[str]) -> bool:
    """Check if a user has all the specified roles.

    Args:
        token: Access token
        role_names: Set of role names to check

    Returns:
        True if user has all the roles, False otherwise
    """
    try:
        user_info = await self.get_userinfo(token)
        if not user_info:
            return False

        # Get all user roles
        all_roles = set()

        # Add realm roles
        realm_access = user_info.get("realm_access", {})
        all_roles.update(realm_access.get("roles", []))

        # Add roles from the configured client
        resource_access = user_info.get("resource_access", {})
        client_roles = resource_access.get(self.configs.CLIENT_ID, {}).get("roles", [])
        all_roles.update(client_roles)

        # Check if all required roles are present
        return role_names.issubset(all_roles)

    except Exception as e:
        logger.debug(f"All roles check failed: {e!s}")
        return False

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.check_permissions async

check_permissions(
    token: str, resource: str, scope: str
) -> bool

Check if a user has permission to access a resource with the specified scope.

Parameters:

Name	Type	Description	Default
token	str	Access token	required
resource	str	Resource name	required
scope	str	Permission scope	required

Returns:

Type	Description
bool	True if permission granted, False otherwise

Source code in archipy/adapters/keycloak/adapters.py

@override
async def check_permissions(self, token: str, resource: str, scope: str) -> bool:
    """Check if a user has permission to access a resource with the specified scope.

    Args:
        token: Access token
        resource: Resource name
        scope: Permission scope

    Returns:
        True if permission granted, False otherwise
    """
    try:
        # Use UMA permissions endpoint to check specific resource and scope
        permissions = await self.openid_adapter.a_uma_permissions(token, permissions=f"{resource}#{scope}")

        # Check if the response indicates permission is granted
        if not permissions or not isinstance(permissions, list):
            logger.debug("No permissions returned or invalid response format")
            return False

        # Look for the specific permission in the response
        for perm in permissions:
            if perm.get("rsname") == resource and scope in perm.get("scopes", []):
                return True

    except KeycloakError as e:
        logger.debug(f"Permission check failed with Keycloak error: {e!s}")
        return False
    except Exception as e:
        logger.debug(f"Permission check failed with unexpected error: {e!s}")
        return False
    else:
        return False

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.create_realm async

create_realm(
    realm_name: str, skip_exists: bool = True, **kwargs: Any
) -> dict[str, Any] | None

Create a Keycloak realm with minimum required fields and optional additional config.

Parameters:

Name	Type	Description	Default
realm_name	str	The realm identifier (required)	required
skip_exists	bool	Skip creation if realm already exists	True
kwargs	Any	Additional optional configurations for the realm	{}

Returns:

Type	Description
dict[str, Any] | None	Dictionary with realm information and status

Raises:

Type	Description
InternalError	If realm creation fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def create_realm(self, realm_name: str, skip_exists: bool = True, **kwargs: Any) -> dict[str, Any] | None:
    """Create a Keycloak realm with minimum required fields and optional additional config.

    Args:
        realm_name: The realm identifier (required)
        skip_exists: Skip creation if realm already exists
        kwargs: Additional optional configurations for the realm

    Returns:
        Dictionary with realm information and status

    Raises:
        InternalError: If realm creation fails
    """
    payload = {
        "realm": realm_name,
        "enabled": kwargs.get("enabled", True),
        "displayName": kwargs.get("display_name", realm_name),
    }

    # Add any additional parameters from kwargs
    for key, value in kwargs.items():
        # Skip display_name as it's already handled
        if key == "display_name":
            continue

        # Convert Python snake_case to Keycloak camelCase
        camel_key = StringUtils.snake_to_camel_case(key)
        payload[camel_key] = value

    try:
        await self.admin_adapter.a_create_realm(payload=payload, skip_exists=skip_exists)
    except KeycloakError as e:
        logger.debug(f"Failed to create realm: {e!s}")

        # Handle realm already exists with skip_exists option
        if skip_exists:
            error_message = self._extract_error_message(e).lower()
            if "already exists" in error_message and "realm" in error_message:
                return {"realm": realm_name, "status": "already_exists", "config": payload}

        # Use the mixin to handle realm-specific errors
        self._handle_realm_exception(e, "create_realm", realm_name)
    else:
        return {"realm": realm_name, "status": "created", "config": payload}

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_realm async

get_realm(realm_name: str) -> dict[str, Any] | None

Get realm details by realm name.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required

Returns:

Type	Description
dict[str, Any] | None	Realm details

Raises:

Type	Description
InternalError	If getting realm fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_realm(self, realm_name: str) -> dict[str, Any] | None:
    """Get realm details by realm name.

    Args:
        realm_name: Name of the realm

    Returns:
        Realm details

    Raises:
        InternalError: If getting realm fails
    """
    try:
        return await self.admin_adapter.a_get_realm(realm_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_realm")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.update_realm async

update_realm(
    realm_name: str, **kwargs: Any
) -> dict[str, Any] | None

Update a realm. Kwargs are RealmRepresentation top-level attributes (e.g. displayName, organizationsEnabled).

Parameters:

Name	Type	Description	Default
realm_name	str	Realm name (not the realm id).	required
**kwargs	Any	RealmRepresentation attributes to update (e.g. displayName, organizationsEnabled).	{}

Returns:

Type	Description
dict[str, Any] | None	Response from Keycloak, or None on error (handled via exception).

Source code in archipy/adapters/keycloak/adapters.py

@override
async def update_realm(self, realm_name: str, **kwargs: Any) -> dict[str, Any] | None:
    """Update a realm. Kwargs are RealmRepresentation top-level attributes (e.g. displayName, organizationsEnabled).

    Args:
        realm_name: Realm name (not the realm id).
        **kwargs: RealmRepresentation attributes to update (e.g. displayName, organizationsEnabled).

    Returns:
        Response from Keycloak, or None on error (handled via exception).
    """
    try:
        return await self.admin_adapter.a_update_realm(realm_name, dict(kwargs))
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "update_realm")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.create_client async

create_client(
    client_id: str,
    realm: str | None = None,
    skip_exists: bool = True,
    **kwargs: Any,
) -> dict[str, Any] | None

Create a Keycloak client with minimum required fields and optional additional config.

Parameters:

Name	Type	Description	Default
client_id	str	The client identifier (required)	required
realm	str | None	Target realm name (uses the current realm in KeycloakAdmin if not specified)	None
skip_exists	bool	Skip creation if client already exists	True
kwargs	Any	Additional optional configurations for the client	{}

Returns:

Type	Description
dict[str, Any] | None	Dictionary with client information

Raises:

Type	Description
InternalError	If client creation fails

Source code in archipy/adapters/keycloak/adapters.py

@override
async def create_client(
    self,
    client_id: str,
    realm: str | None = None,
    skip_exists: bool = True,
    **kwargs: Any,
) -> dict[str, Any] | None:
    """Create a Keycloak client with minimum required fields and optional additional config.

    Args:
        client_id: The client identifier (required)
        realm: Target realm name (uses the current realm in KeycloakAdmin if not specified)
        skip_exists: Skip creation if client already exists
        kwargs: Additional optional configurations for the client

    Returns:
        Dictionary with client information

    Raises:
        InternalError: If client creation fails
    """
    original_realm = self.admin_adapter.connection.realm_name

    try:
        # Set the target realm if provided
        if realm and realm != original_realm:
            self.admin_adapter.connection.realm_name = realm

        public_client = kwargs.get("public_client", False)

        # Prepare the minimal client payload
        payload = {
            "clientId": client_id,
            "enabled": kwargs.get("enabled", True),
            "protocol": kwargs.get("protocol", "openid-connect"),
            "name": kwargs.get("name", client_id),
            "publicClient": public_client,
        }

        # Enable service accounts for confidential clients by default
        if not public_client:
            payload["serviceAccountsEnabled"] = kwargs.get("service_account_enabled", True)
            payload["clientAuthenticatorType"] = "client-secret"

        for key, value in kwargs.items():
            if key in ["enabled", "protocol", "name", "public_client", "service_account_enabled"]:
                continue

            # Convert snake_case to camelCase
            camel_key = StringUtils.snake_to_camel_case(key)
            payload[camel_key] = value

        internal_client_id = None
        try:
            internal_client_id = await self.admin_adapter.a_create_client(payload, skip_exists=skip_exists)
        except KeycloakError as e:
            logger.debug(f"Failed to create client: {e!s}")

            # Handle client already exists with skip_exists option
            if skip_exists:
                error_message = self._extract_error_message(e).lower()
                if "already exists" in error_message and "client" in error_message:
                    return {
                        "client_id": client_id,
                        "status": "already_exists",
                        "realm": self.admin_adapter.connection.realm_name,
                    }

            # Use the mixin to handle client-specific errors
            client_data = {"clientId": client_id, "name": kwargs.get("name", client_id)}
            self._handle_client_exception(e, "create_client", client_data)

        return {
            "client_id": client_id,
            "internal_client_id": internal_client_id,
            "realm": self.admin_adapter.connection.realm_name,
            "status": "created",
        }

    finally:
        # Always restore the original realm
        if realm and realm != original_realm:
            self.admin_adapter.connection.realm_name = original_realm

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.add_realm_roles_to_composite async

add_realm_roles_to_composite(
    composite_role_name: str, child_role_names: list[str]
) -> None

Add realm roles to a composite role.

Parameters:

Name	Type	Description	Default
composite_role_name	str	Name of the composite role	required
child_role_names	list[str]	List of child role names to add	required

Source code in archipy/adapters/keycloak/adapters.py

@override
async def add_realm_roles_to_composite(self, composite_role_name: str, child_role_names: list[str]) -> None:
    """Add realm roles to a composite role.

    Args:
        composite_role_name: Name of the composite role
        child_role_names: List of child role names to add
    """
    try:
        child_roles = []
        for role_name in child_role_names:
            try:
                role = await self.admin_adapter.a_get_realm_role(role_name)
                child_roles.append(role)
            except KeycloakGetError as e:
                if e.response_code == 404:
                    logger.warning(f"Child role not found: {role_name}")
                    continue
                raise

        if child_roles:
            await self.admin_adapter.a_add_composite_realm_roles_to_role(
                role_name=composite_role_name,
                roles=child_roles,
            )
            logger.info(f"Added {len(child_roles)} realm roles to composite role: {composite_role_name}")

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "add_realm_roles_to_composite")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.add_client_roles_to_composite async

add_client_roles_to_composite(
    composite_role_name: str,
    client_id: str,
    child_role_names: list[str],
) -> None

Add client roles to a composite role.

Parameters:

Name	Type	Description	Default
composite_role_name	str	Name of the composite role	required
client_id	str	Client ID or client name	required
child_role_names	list[str]	List of child role names to add	required

Source code in archipy/adapters/keycloak/adapters.py

@override
async def add_client_roles_to_composite(
    self,
    composite_role_name: str,
    client_id: str,
    child_role_names: list[str],
) -> None:
    """Add client roles to a composite role.

    Args:
        composite_role_name: Name of the composite role
        client_id: Client ID or client name
        child_role_names: List of child role names to add
    """
    try:
        internal_client_id = await self.admin_adapter.a_get_client_id(client_id)
        if internal_client_id is None:
            raise ValueError("client_id resolved to None")

        child_roles = []
        for role_name in child_role_names:
            try:
                # Keycloak admin adapter methods accept these types at runtime
                role = await self.admin_adapter.a_get_client_role(internal_client_id, role_name)
                child_roles.append(role)
            except KeycloakGetError as e:
                if e.response_code == 404:
                    logger.warning(f"Client role not found: {role_name}")
                    continue
                raise

        if child_roles:
            if internal_client_id is None:
                raise ValueError("Client ID not found")
            resolved_client_id: str = internal_client_id
            await self.admin_adapter.a_add_composite_client_roles_to_role(
                role_name=composite_role_name,
                client_role_id=resolved_client_id,
                roles=child_roles,
            )
            logger.info(f"Added {len(child_roles)} client roles to composite role: {composite_role_name}")

    except KeycloakError as e:
        self._handle_keycloak_exception(e, "add_client_roles_to_composite")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_composite_realm_roles async

get_composite_realm_roles(
    role_name: str,
) -> list[dict[str, Any]] | None

Get composite roles for a realm role.

Parameters:

Name	Type	Description	Default
role_name	str	Name of the role	required

Returns:

Type	Description
list[dict[str, Any]] | None	List of composite roles

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_composite_realm_roles(self, role_name: str) -> list[dict[str, Any]] | None:
    """Get composite roles for a realm role.

    Args:
        role_name: Name of the role

    Returns:
        List of composite roles
    """
    try:
        return await self.admin_adapter.a_get_composite_realm_roles_of_role(role_name)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_composite_realm_roles")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_organizations async

get_organizations(
    query: dict | None = None,
) -> list[dict[str, Any]]

Fetch all organizations, optionally filtered by query parameters.

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_organizations(self, query: dict | None = None) -> list[dict[str, Any]]:
    """Fetch all organizations, optionally filtered by query parameters."""
    try:
        return await self.admin_adapter.a_get_organizations(query=query)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_organizations")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_organization async

get_organization(organization_id: str) -> dict[str, Any]

Get representation of the organization by ID.

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_organization(self, organization_id: str) -> dict[str, Any]:
    """Get representation of the organization by ID."""
    try:
        return await self.admin_adapter.a_get_organization(organization_id=organization_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_organization")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.create_organization async

create_organization(
    name: str, alias: str, **kwargs: Any
) -> str | None

Create a new organization. Name and alias must be unique. Returns org_id.

Source code in archipy/adapters/keycloak/adapters.py

@override
async def create_organization(self, name: str, alias: str, **kwargs: Any) -> str | None:
    """Create a new organization. Name and alias must be unique. Returns org_id."""
    try:
        payload = {"name": name, "alias": alias}
        for key, value in kwargs.items():
            if key in ["name", "alias"]:
                continue

            # Convert snake_case to camelCase
            camel_key = StringUtils.snake_to_camel_case(key)
            payload[camel_key] = value

        return await self.admin_adapter.a_create_organization(payload=payload)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "create_organization")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.update_organization async

update_organization(
    organization_id: str, **kwargs: Any
) -> dict[str, Any]

Update an existing organization. Kwargs are organization attributes (e.g. name, alias).

Source code in archipy/adapters/keycloak/adapters.py

@override
async def update_organization(self, organization_id: str, **kwargs: Any) -> dict[str, Any]:
    """Update an existing organization. Kwargs are organization attributes (e.g. name, alias)."""
    try:
        payload = {}
        for key, value in kwargs.items():
            # Convert snake_case to camelCase
            camel_key = StringUtils.snake_to_camel_case(key)
            payload[camel_key] = value

        return await self.admin_adapter.a_update_organization(organization_id=organization_id, payload=payload)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "update_organization")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.delete_organization async

delete_organization(organization_id: str) -> dict[str, Any]

Delete an organization.

Source code in archipy/adapters/keycloak/adapters.py

@override
async def delete_organization(self, organization_id: str) -> dict[str, Any]:
    """Delete an organization."""
    try:
        return await self.admin_adapter.a_delete_organization(organization_id=organization_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "delete_organization")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_user_organizations async

get_user_organizations(
    user_id: str,
) -> list[dict[str, Any]]

Get organizations by user id.

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_user_organizations(self, user_id: str) -> list[dict[str, Any]]:
    """Get organizations by user id."""
    try:
        return await self.admin_adapter.a_get_user_organizations(user_id=user_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_user_organizations")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_organization_members async

get_organization_members(
    organization_id: str, query: dict | None = None
) -> list[dict[str, Any]]

Get members by organization id, optionally filtered by query parameters.

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_organization_members(self, organization_id: str, query: dict | None = None) -> list[dict[str, Any]]:
    """Get members by organization id, optionally filtered by query parameters."""
    try:
        return await self.admin_adapter.a_get_organization_members(organization_id=organization_id, query=query)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_organization_members")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_organization_members_count async

get_organization_members_count(organization_id: str) -> int

Get the number of members in the organization.

Source code in archipy/adapters/keycloak/adapters.py

@override
async def get_organization_members_count(self, organization_id: str) -> int:
    """Get the number of members in the organization."""
    try:
        return await self.admin_adapter.a_get_organization_members_count(organization_id=organization_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "get_organization_members_count")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.organization_user_add async

organization_user_add(
    user_id: str, organization_id: str
) -> bytes

Add a user to an organization.

Source code in archipy/adapters/keycloak/adapters.py

@override
async def organization_user_add(self, user_id: str, organization_id: str) -> bytes:
    """Add a user to an organization."""
    try:
        return await self.admin_adapter.a_organization_user_add(user_id=user_id, organization_id=organization_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "organization_user_add")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.organization_user_remove async

organization_user_remove(
    user_id: str, organization_id: str
) -> dict[str, Any]

Remove a user from an organization.

Source code in archipy/adapters/keycloak/adapters.py

@override
async def organization_user_remove(self, user_id: str, organization_id: str) -> dict[str, Any]:
    """Remove a user from an organization."""
    try:
        return await self.admin_adapter.a_organization_user_remove(user_id=user_id, organization_id=organization_id)
    except KeycloakError as e:
        self._handle_keycloak_exception(e, "organization_user_remove")

archipy.adapters.keycloak.adapters.AsyncKeycloakAdapter.get_organization_idps abstractmethod async

get_organization_idps(
    organization_id: str,
) -> list[dict[str, Any]]

Get IDPs by organization id.

Source code in archipy/adapters/keycloak/ports.py

@abstractmethod
async def get_organization_idps(self, organization_id: str) -> list[dict[str, Any]]:
    """Get IDPs by organization id."""
    raise NotImplementedError

options: show_root_toc_entry: false heading_level: 3