API Reference

This section documents the public Python modules of the operator.

Operator Entry Point

Keycloak Operator - Main entry point for the Kopf-based Keycloak operator.

This operator provides GitOps-compatible Keycloak management with: - Multi-namespace operation (watches all namespaces by default) - Dynamic client provisioning with RBAC-based authorization - Kubernetes-native security instead of Keycloak's built-in auth - Comprehensive secret management improvements

Usage

python -m keycloak_operator.operator

Or with kopf directly:

kopf run keycloak_operator.operator --verbose --all-namespaces

Environment Variables

KEYCLOAK_OPERATOR_NAMESPACES: Comma-separated list of namespaces to watch KEYCLOAK_OPERATOR_LOG_LEVEL: Logging level (DEBUG, INFO, WARNING, ERROR) KEYCLOAK_OPERATOR_DRY_RUN: Set to 'true' for dry-run mode

_setup_coverage_signal_handler

_setup_coverage_signal_handler()

Setup SIGUSR1 signal handler to flush coverage data on demand.

This allows graceful coverage data collection during integration tests by: 1. Sending SIGUSR1 to the operator process 2. Handler calls coverage.save() to flush data to disk 3. Coverage files can then be retrieved while operator continues running

Only enabled when COVERAGE_PROCESS_START environment variable is set.

Raises:

Type	Description
ImportError	If coverage module is not installed when coverage is enabled

cleanup_handler async

cleanup_handler(settings, **_)

Operator cleanup handler.

This runs when the operator is shutting down and can be used to perform cleanup tasks like: - Closing database connections - Cleaning up temporary resources - Stopping metrics server - Logging shutdown information

configure_logging

configure_logging()

Configure structured logging for the operator based on operator_settings.

drift_detection_timer async

drift_detection_timer(**kwargs)

Periodic drift detection task.

This task runs on a timer to check for drift between Keycloak state and Kubernetes CRs. It detects: - Orphaned resources (created by operator but CR deleted) - Configuration drift (CR exists but state differs) - Unmanaged resources (exist in Keycloak without operator ownership)

get_watched_namespaces

get_watched_namespaces()

Get the list of namespaces to watch from operator_settings.

Returns:

Type	Description
list[str] | None	List of namespace names, or None to watch all namespaces

health_check async

health_check(**_)

Health check probe for Kubernetes liveness/readiness checks.

Returns:

Type	Description
dict[str, str]	Dictionary indicating operator health status

main

main()

Main entry point for the operator.

This function: 1. Configures logging 2. Determines namespace scope 3. Configures admission webhooks (must be before kopf.run()) 4. Runs the kopf operator with appropriate settings

readiness_check async

readiness_check(**_)

Readiness check probe - indicates if operator is ready to handle requests.

Returns:

Type	Description
dict[str, str]	Dictionary indicating operator readiness

startup_handler async

startup_handler(settings, memo, **_)

Operator startup configuration.

This handler runs once when the operator starts up and configures global operator settings including: - Resource scanning behavior - Error handling policies - Networking settings - Performance tuning - Metrics and health check endpoints - Rate limiting for Keycloak API calls - Admission webhook server

Models

Pydantic models for Keycloak instance resources.

This module defines type-safe data models for Keycloak instance specifications and status. These models ensure proper validation and provide IDE support for the operator development.

ConnectionPoolConfig

Bases: BaseModel

Database connection pool configuration.

Keycloak

Bases: BaseModel

Complete Keycloak custom resource model.

This represents the full Kubernetes custom resource including metadata, spec, and status sections.

KeycloakCondition

Bases: BaseModel

Status condition for Keycloak instance.

KeycloakDatabaseConfig

Bases: BaseModel

Database configuration for Keycloak instance.

Production-ready configuration that enforces external database usage. For CloudNativePG clusters, use standard PostgreSQL connection details.

validate_database_configuration

validate_database_configuration()

Validate complete database configuration with production-ready requirements.

KeycloakEndpoints

Bases: BaseModel

Endpoints for accessing Keycloak instance.

KeycloakIngressConfig

Bases: BaseModel

Ingress configuration for Keycloak instance.

KeycloakInstanceRef

Bases: BaseModel

Reference to a Keycloak instance in any namespace.

KeycloakResourceRequirements

Bases: BaseModel

Resource requirements for Keycloak pods.

KeycloakServiceConfig

Bases: BaseModel

Service configuration for Keycloak instance.

KeycloakSpec

Bases: BaseModel

Specification for a Keycloak instance.

This model defines all configurable aspects of a Keycloak deployment including resources, networking, persistence, and authentication.

KeycloakStatus

Bases: BaseModel

Status of a Keycloak instance.

This model represents the current state and health of a Keycloak deployment as observed by the operator.

KeycloakTLSConfig

Bases: BaseModel

TLS configuration for Keycloak instance.

KeycloakTracingConfig

Bases: BaseModel

OpenTelemetry distributed tracing configuration for Keycloak.

Keycloak 26.x+ has built-in OpenTelemetry support via Quarkus. This configuration enables end-to-end distributed tracing.

RealmCapacity

Bases: BaseModel

Capacity management for realms.

Controls how many realms can be managed by this Keycloak operator and whether new realm creation is allowed.

SecretReference

Bases: BaseModel

Reference to a secret key for sensitive data.

The secret must be in the same namespace as the resource referencing it. Cross-namespace secret references are not supported for security reasons.

Pydantic models for KeycloakRealm resources.

This module defines type-safe data models for Keycloak realm specifications and status. These models enable comprehensive realm management including authentication flows, identity providers, and user federation.

AuthenticationExecutionExport

Bases: BaseModel

Authentication execution step configuration.

Maps to AuthenticationExecutionExportRepresentation in Keycloak API. Represents a single step in an authentication flow.

model_post_init

model_post_init(__context)

Validate that execution has either authenticator or flow reference.

AuthenticatorConfigInfo

Bases: BaseModel

Configuration for a configurable authenticator execution.

Maps to AuthenticatorConfigRepresentation in Keycloak API. Used for authenticators that have configuration options (e.g., OTP settings).

ClientPolicy

Bases: BaseModel

Client policy that conditionally applies profiles to clients.

Policies define WHEN profiles should be applied (via conditions) and WHICH profiles to apply. When all conditions match, all referenced profiles' executors are enforced on the client.

Example YAML

clientPolicies: - name: enforce-fapi-for-confidential description: Apply FAPI profile to all confidential clients enabled: true conditions: - condition: client-access-type configuration: type: - confidential profiles: - fapi-advanced

ClientPolicyCondition

Bases: BaseModel

Condition that determines when a client policy applies.

Conditions evaluate client attributes to decide if the policy's profiles should be applied. Multiple conditions in a policy are ANDed together.

Common condition types: - client-roles: Match clients with specific roles - client-scopes: Match clients with specific scopes - client-access-type: Match by access type (public/confidential) - any-client: Match all clients

Example

ClientPolicyCondition( condition="client-access-type", configuration={"type": ["confidential"]} )

ClientProfile

Bases: BaseModel

Client profile defining a set of executors applied to matching clients.

Client profiles group multiple executors together under a single name. Profiles are referenced by client policies to apply their executors when policy conditions match.

Example YAML

clientProfiles: - name: fapi-advanced description: FAPI 2.0 Advanced Security Profile executors: - executor: pkce-enforcer configuration: auto-configure: "true" - executor: secure-client-authenticator configuration: allowed-client-authenticators: - private_key_jwt

ClientProfileExecutor

Bases: BaseModel

Executor configuration for a client profile.

Executors enforce specific behaviors on clients, such as: - secure-client-authenticator: Require specific authentication methods - pkce-enforcer: Enforce PKCE for authorization code flow - consent-required: Require user consent - secure-redirect-uris-enforcer: Validate redirect URIs

Example

ClientProfileExecutor( executor="pkce-enforcer", configuration={"auto-configure": "true"} )

KeycloakAuthenticationFlow

Bases: BaseModel

Authentication flow configuration.

Maps to AuthenticationFlowRepresentation in Keycloak API. Defines a complete authentication flow with its executions.

KeycloakClientScope

Bases: BaseModel

Client scope definition.

KeycloakEventsConfig

Bases: BaseModel

Event logging configuration.

KeycloakGroup

Bases: BaseModel

Group definition.

KeycloakIdentityProvider

Bases: BaseModel

Identity provider configuration.

model_post_init

model_post_init(__context)

Validate that sensitive keys are only in configSecrets, not plaintext in config.

KeycloakIdentityProviderMapper

Bases: BaseModel

Identity provider mapper configuration.

Mappers transform claims/attributes from the identity provider into Keycloak user attributes, roles, or session attributes.

KeycloakIdentityProviderSecretRef

Bases: BaseModel

Reference to Kubernetes secret containing identity provider secrets.

The secret must be in the same namespace as the KeycloakRealm. Cross-namespace secret references are not supported for security reasons.

KeycloakOTPPolicy

Bases: BaseModel

OTP (One-Time Password) policy configuration.

Maps to flat otpPolicy* fields in Keycloak RealmRepresentation.

KeycloakPasswordPolicy

Bases: BaseModel

Password policy configuration for a realm.

Enforces password requirements for users. The policy is converted to Keycloak's "and"-separated policy string format.

to_policy_string

to_policy_string()

Convert to Keycloak password policy string format.

Returns:

Type	Description
str	Policy string with " and " separator (e.g., "length(12) and upperCase(1) and notUsername")

KeycloakProtocolMapper

Bases: BaseModel

Protocol mapper for client scopes.

KeycloakRealm

Bases: BaseModel

Complete KeycloakRealm custom resource model.

This represents the full Kubernetes custom resource for comprehensive realm management with authentication flows and identity providers.

KeycloakRealmCondition

Bases: BaseModel

Status condition for KeycloakRealm resource.

KeycloakRealmEndpoints

Bases: BaseModel

Endpoints for the KeycloakRealm.

KeycloakRealmFeatures

Bases: BaseModel

Features configured for the realm.

KeycloakRealmLocalization

Bases: BaseModel

Localization configuration for a realm.

KeycloakRealmRole

Bases: BaseModel

Realm role definition.

KeycloakRealmSecurity

Bases: BaseModel

Security settings for a realm.

KeycloakRealmSpec

Bases: BaseModel

Specification for a KeycloakRealm resource.

This model defines all configurable aspects of a Keycloak realm including security, authentication, and user management settings.

to_keycloak_config

to_keycloak_config(include_flow_bindings=True)

Convert the realm specification to Keycloak API format.

Parameters:

Name	Type	Description	Default
include_flow_bindings	bool	Whether to include flow binding fields. Set to False when creating realm initially (flows don't exist yet).	True

Returns:

Type	Description
dict[str, Any]	Dictionary in Keycloak Admin API format

validate_namespace_grants classmethod

validate_namespace_grants(v)

Validate namespace names in authorization grants.

KeycloakRealmStatus

Bases: BaseModel

Status of a KeycloakRealm resource.

This model represents the current state of a realm as managed by the operator.

KeycloakRealmTheme

Bases: BaseModel

Theme configuration for a realm.

KeycloakRealmTokenSettings

Bases: BaseModel

Token settings for a realm.

KeycloakRoles

Bases: BaseModel

Realm and client role definitions.

KeycloakSMTPConfig

Bases: BaseModel

SMTP server configuration with validation.

model_post_init

model_post_init(__context)

Validate auth requirements after model construction.

validate_auth_requirements classmethod

validate_auth_requirements(v, info)

Ensure auth settings are consistent.

KeycloakSMTPPasswordSecret

Bases: BaseModel

Reference to Kubernetes secret containing SMTP password.

The secret must be in the same namespace as the KeycloakRealm. Cross-namespace secret references are not supported for security reasons.

KeycloakUserFederation

Bases: BaseModel

Complete user federation configuration for LDAP/AD/Kerberos.

This model supports: - LDAP federation with full attribute mapping - Active Directory with sAMAccountName/UPN support - Kerberos/SPNEGO authentication - User/group synchronization

Example LDAP configuration

KeycloakUserFederation( name="corporate-ldap", provider_id="ldap", connection_url="ldap://ldap.example.com:389", users_dn="ou=People,dc=example,dc=org", bind_dn="cn=admin,dc=example,dc=org", bind_credential_secret=KeycloakUserFederationSecretRef( name="ldap-credentials", key="password" ), )

to_component_config

to_component_config()

Convert to Keycloak Components API config format.

Keycloak's Components API expects config values as lists of strings.

Returns:

Type	Description
dict[str, list[str]]	Dictionary mapping config keys to list of string values

KeycloakUserFederationMapper

Bases: BaseModel

User federation mapper configuration.

Mappers transform LDAP/AD attributes into Keycloak user properties, roles, or group memberships.

KeycloakUserFederationSecretRef

Bases: BaseModel

Reference to Kubernetes secret for user federation sensitive data.

Used for bind credentials (LDAP) or keytab (Kerberos).

KeycloakUserFederationStatus

Bases: BaseModel

Status of a user federation provider.

This provides observability into the health and sync state of each configured user federation provider (LDAP, AD, Kerberos).

KeycloakUserFederationSyncSettings

Bases: BaseModel

Synchronization settings for user federation.

KeycloakWebAuthnPasswordlessPolicy

Bases: KeycloakWebAuthnPolicy

WebAuthn Passwordless policy configuration.

Inherits all fields from KeycloakWebAuthnPolicy.

KeycloakWebAuthnPolicy

Bases: BaseModel

WebAuthn policy configuration.

Used for both regular WebAuthn and Passwordless WebAuthn policies.

OperatorRef

Bases: BaseModel

Reference to the operator managing this realm.

Organization

Bases: BaseModel

Organization definition for multi-tenancy support (Keycloak 26+).

Organizations provide a way to group users and resources for multi-tenant applications. Each organization can have its own members, domains, and identity providers.

IMPORTANT: Organizations require Keycloak version 26.0.0 or higher. The operator will validate the Keycloak version before attempting to create organizations.

Example YAML

organizations: - name: acme-corp alias: acme description: ACME Corporation enabled: true domains: - name: acme.com verified: true attributes: industry: ["technology"] tier: ["enterprise"]

OrganizationDomain

Bases: BaseModel

Domain configuration for an organization.

Domains are used to identify members of an organization based on their email domain. Members with emails matching verified domains can be automatically associated with the organization.

OrganizationIdentityProvider

Bases: BaseModel

Identity provider linked to an organization.

Organizations can have their own identity providers that members use for authentication. This enables federated identity for organization members.

RequiredActionProvider

Bases: BaseModel

Required action configuration.

Maps to RequiredActionProviderRepresentation in Keycloak API. Required actions are actions users must perform (e.g., verify email, configure OTP).

Pydantic models for KeycloakClient resources.

This module defines type-safe data models for Keycloak client specifications and status. These models enable dynamic client provisioning across namespaces with proper validation and GitOps compatibility.

AggregatePolicy

Bases: BaseModel

Aggregate authorization policy.

Combines multiple policies using a decision strategy.

AuthorizationPermissions

Bases: BaseModel

Container for all authorization permission types.

Permissions tie policies to resources/scopes to create the complete authorization model. They define WHAT can be accessed and link to policies that define WHO can access.

AuthorizationPolicies

Bases: BaseModel

Container for all authorization policy types.

Policies define WHO can access resources. They are referenced by permissions to create the complete authorization model.

AuthorizationResource

Bases: BaseModel

Protected resource definition.

Resources represent the objects being protected by authorization policies (e.g., APIs, documents, users).

AuthorizationScope

Bases: BaseModel

Authorization scope definition.

Scopes define the actions that can be performed on resources (e.g., 'read', 'write', 'delete').

AuthorizationSettings

Bases: BaseModel

Authorization services settings for a client.

Contains the resource server configuration including enforcement mode, decision strategy, scopes, and resources.

ClientPolicy

Bases: BaseModel

Client-based authorization policy.

Grants access to specific OAuth2 clients.

GroupPolicy

Bases: BaseModel

Group-based authorization policy.

Grants access based on group membership.

JavaScriptPolicy

Bases: BaseModel

JavaScript-based authorization policy.

WARNING: JavaScript policies require the 'upload-scripts' feature to be enabled in Keycloak, which is disabled by default for security reasons. Use with caution.

The policy code has access to evaluation context with user, client, and resource info.

KeycloakClient

Bases: BaseModel

Complete KeycloakClient custom resource model.

This represents the full Kubernetes custom resource for dynamic client provisioning with cross-namespace support.

KeycloakClientAuthenticationFlow

Bases: BaseModel

Configuration for client authentication flows.

KeycloakClientCondition

Bases: BaseModel

Status condition for KeycloakClient resource.

KeycloakClientEndpoints

Bases: BaseModel

Endpoints for the KeycloakClient.

KeycloakClientProtocolMapper

Bases: BaseModel

Configuration for protocol mappers.

KeycloakClientScope

Bases: BaseModel

Configuration for client scopes.

KeycloakClientSecretRef

Bases: BaseModel

Reference to an existing Kubernetes secret containing the client secret.

Used when manually managing the client secret instead of letting the operator generate one.

KeycloakClientSettings

Bases: BaseModel

Advanced client settings.

KeycloakClientSpec

Bases: BaseModel

Specification for a KeycloakClient resource.

This model defines all configurable aspects of a Keycloak client including authentication, authorization, and protocol settings.

to_keycloak_config

to_keycloak_config()

Convert the client specification to Keycloak API format.

Returns:

Type	Description
dict[str, Any]	Dictionary in Keycloak Admin API format

validate_redirect_uris classmethod

validate_redirect_uris(v)

Validate redirect URIs follow Keycloak wildcard rules.

Keycloak allows wildcards () in specific locations: - ✓ In path: http://localhost:3000/ or https://example.com/app/ - ✓ Custom schemes: custom: or mycustomscheme: - ✗ In domain: https://.example.com or http://example*.com - ✗ Bare wildcard: * (too permissive, blocked since Keycloak 22.x)

Wildcards can only appear at the END of the URI.

validate_secret_configuration

validate_secret_configuration()

Validate secret configuration constraints.

KeycloakClientStatus

Bases: BaseModel

Status of a KeycloakClient resource.

This model represents the current state of a client as managed by the operator.

RealmRef

Bases: BaseModel

Reference to a parent KeycloakRealm.

RegexPolicy

Bases: BaseModel

Regex-based authorization policy.

Grants access based on regex matching against a token claim.

ResourcePermission

Bases: BaseModel

Resource-based authorization permission.

Defines access rights for specific resources with associated policies. This permission type grants access to entire resources (all scopes).

RolePolicy

Bases: BaseModel

Role-based authorization policy.

Grants access based on realm or client role assignments.

RolePolicyRole

Bases: BaseModel

A role reference for role-based policies.

ScopePermission

Bases: BaseModel

Scope-based authorization permission.

Defines access rights for specific scopes, optionally on specific resources. This permission type provides finer-grained control than resource permissions.

SecretMetadata

Bases: BaseModel

Metadata to be added to the managed secret.

SecretRotationConfig

Bases: BaseModel

Configuration for automated client secret rotation.

validate_rotation_time classmethod

validate_rotation_time(v)

Validate rotation_time is in HH:MM format with valid hour/minute values.

ServiceAccountRoles

Bases: BaseModel

Role mappings for service account users.

TimePolicy

Bases: BaseModel

Time-based authorization policy.

Grants access based on time constraints. All time fields are optional - only specified constraints are enforced.

UserPolicy

Bases: BaseModel

User-based authorization policy.

Grants access to specific users by username.

Handlers

Keycloak instance handlers - Manages the core Keycloak deployment and services.

This module handles the lifecycle of Keycloak instances including: - Creating Keycloak deployments with proper configuration - Managing services and ingress for external access - Setting up persistent storage for Keycloak data - Configuring initial admin users and realms - Health monitoring and status reporting

The handlers in this module are designed to be idempotent and GitOps-friendly, ensuring that the desired state is maintained regardless of restart or failure.

KopfHandlerKwargs

Bases: TypedDict

Type hints for common kopf handler kwargs.

StatusProtocol

Bases: Protocol

Protocol for kopf Status objects that allow dynamic attribute assignment.

Wrapped by StatusWrapper to allow safe mutation irrespective of kopf internal status object semantics.

StatusWrapper

StatusWrapper(patch_status)

Bases: MutableMapping[str, Any]

Safe mutable wrapper around kopf patch.status for both item & attribute access.

Automatically converts snake_case Python attribute names to camelCase for K8s API.

_to_camel_case staticmethod

_to_camel_case(snake_str)

Convert snake_case to camelCase.

_perform_keycloak_cleanup async

_perform_keycloak_cleanup(
    name,
    namespace,
    spec,
    rate_limiter,
    trigger="delete_handler",
)

Perform Keycloak instance cleanup.

This is the core cleanup logic for delete handlers. Finalizer management is handled by Kopf via settings.persistence.finalizer.

Parameters:

Name	Type	Description	Default
name	str	Name of the Keycloak resource	required
namespace	str	Namespace where the resource exists	required
spec	dict[str, Any]	Keycloak resource specification	required
rate_limiter	Any	Rate limiter for API calls	required
trigger	str	What triggered this cleanup (for logging)	'delete_handler'

_record_instance_status

_record_instance_status(namespace, running)

Record Keycloak instance status to Prometheus metrics.

_run_keycloak_health_check async

_run_keycloak_health_check(
    spec, name, namespace, status, patch, meta, memo
)

Execute one health check iteration for a Keycloak instance.

delete_keycloak_instance async

delete_keycloak_instance(
    spec,
    name,
    namespace,
    status,
    patch,
    memo,
    retry,
    **kwargs
)

Handle Keycloak instance deletion.

This handler performs comprehensive cleanup of all associated resources. Finalizer management is handled automatically by Kopf via settings.persistence.finalizer.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	Keycloak resource specification	required
name	str	Name of the Keycloak resource	required
namespace	str	Namespace where the resource exists	required
status	StatusProtocol	Current status of the resource	required
patch	Patch	Kopf patch object for modifying the resource	required
retry	int	Kopf retry count (starts from 0)	required

ensure_keycloak_instance async

ensure_keycloak_instance(
    spec, name, namespace, status, patch, memo, **kwargs
)

Ensure Keycloak instance exists and is properly configured.

This is the main handler for Keycloak instance creation and resumption. It implements idempotent logic that works for both initial creation and operator restarts (resume).

Note: Deletion is handled by the @kopf.on.delete handler (delete_keycloak_instance). Do not add deletion logic here to avoid race conditions with the delete handler.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	Keycloak resource specification	required
name	str	Name of the Keycloak resource	required
namespace	str	Namespace where the resource exists	required
status	StatusProtocol	Current status of the resource	required
patch	Patch	Kopf patch object for modifying the resource	required

Returns:

Type	Description
None	Dictionary with status information for the resource

monitor_keycloak_health async

monitor_keycloak_health(
    stopped,
    spec,
    name,
    namespace,
    status,
    patch,
    meta,
    memo,
    **kwargs
)

Periodic health check daemon for Keycloak instances.

This daemon checks the health of Keycloak instances and updates
their status accordingly.

Uses a daemon instead of a timer to work around Kopf 1.40.x not
supporting callable ``initial_delay`` (lambda). The daemon sleeps
a random jitter on startup to spread reconciliation load, then
loops at TIMER_INTERVAL_KEYCLOAK between iterations.

The interval is configurable via TIMER_INTERVAL_KEYCLOAK environment variable.
Default: 600 seconds (10 minutes).

Args:
    stopped: Kopf daemon stopped flag for graceful shutdown
    spec: Keycloak resource specification
    name: Name of the Keycloak resource
    namespace: Namespace where the resource exists
    status: Current status of the resource
    meta: Resource metadata
    memo: Kopf memo for accessing shared state like rate_limiter

Implementation includes

✅ Check if Keycloak deployment is running and ready ✅ Verify that Keycloak is responding to health checks ✅ Check resource utilization (CPU, memory, storage) ✅ Validate that Keycloak Admin API is accessible and master realm exists ✅ Update status with current health information ⚠️ Generate events for significant status changes - Future enhancement ⚠️ Implement alerting for persistent failures - Future enhancement

update_keycloak_instance async

update_keycloak_instance(
    old,
    new,
    diff,
    name,
    namespace,
    status,
    patch,
    memo,
    **kwargs
)

Handle updates to Keycloak instance specifications.

This handler is called when the Keycloak resource specification changes. It delegates to the reconciler service layer for all business logic.

Parameters:

Name	Type	Description	Default
old	dict[str, Any]	Previous specification	required
new	dict[str, Any]	New specification	required
diff	Diff	List of changes between old and new	required
name	str	Name of the Keycloak resource	required
namespace	str	Namespace where the resource exists	required
status	StatusProtocol	Current status of the resource	required
patch	Patch	Kopf patch object for modifying the resource	required

Returns:

Type	Description
dict[str, Any] | None	None to avoid Kopf creating status subpaths

KeycloakRealm handlers - Manages realm lifecycle and configuration.

This module handles realm management within Keycloak instances, including: - Creating and configuring realms - Setting up authentication flows and identity providers - Managing realm-level settings and policies - Configuring user federation and storage - Setting up realm-specific themes and localization

Realms provide isolation between different applications or tenants and can be managed independently across different namespaces.

StatusProtocol

Bases: Protocol

Protocol for kopf Status objects that allow dynamic attribute assignment.

StatusWrapper

StatusWrapper(patch_status)

Wrapper to make kopf patch.status compatible with StatusProtocol.

This wrapper provides both attribute and dict-like access to patch.status, ensuring all updates are written directly to the underlying patch object.

Automatically converts snake_case Python attribute names to camelCase for K8s API.

_to_camel_case staticmethod

_to_camel_case(snake_str)

Convert snake_case to camelCase.

update

update(data)

Update multiple fields. Assumes data keys are already in camelCase.

_perform_realm_cleanup async

_perform_realm_cleanup(
    name,
    namespace,
    spec,
    status,
    rate_limiter,
    trigger="delete_handler",
)

Perform realm cleanup from Keycloak.

This is the core cleanup logic for delete handlers. Finalizer management is handled by Kopf via settings.persistence.finalizer.

Parameters:

Name	Type	Description	Default
name	str	Name of the KeycloakRealm resource	required
namespace	str	Namespace where the resource exists	required
spec	dict[str, Any]	KeycloakRealm resource specification	required
status	dict[str, Any]	Current status of the resource	required
rate_limiter	Any	Rate limiter for Keycloak API calls	required
trigger	str	What triggered this cleanup (for logging)	'delete_handler'

_record_federation_status

_record_federation_status(
    realm_name, provider_id, connected
)

Record user federation provider status to Prometheus metrics.

_run_realm_health_check async

_run_realm_health_check(
    spec, name, namespace, status, patch, meta, memo
)

Execute one health check iteration for a KeycloakRealm.

_test_user_federation async

_test_user_federation(
    admin_client, realm_name, namespace, federation_specs
)

Test user federation connections.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	required
federation_specs	list	List of user federation specifications	required

Returns:

Type	Description
bool	True if federation is healthy, False otherwise

_verify_authentication_flows async

_verify_authentication_flows(
    admin_client, realm_name, namespace, flow_specs
)

Verify that authentication flows exist and are configured correctly.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	required
flow_specs	list	List of expected authentication flow specifications	required

Returns:

Type	Description
bool	True if flows are valid, False otherwise

_verify_identity_providers async

_verify_identity_providers(
    admin_client, realm_name, namespace, idp_specs
)

Verify that identity providers exist and are configured correctly.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	required
idp_specs	list	List of expected identity provider specifications	required

Returns:

Type	Description
bool	True if identity providers are valid, False otherwise

_verify_realm_config

_verify_realm_config(current_realm, realm_spec)

Verify that the current realm configuration matches the spec.

Parameters:

Name	Type	Description	Default
current_realm	dict[str, Any]	Current realm configuration from Keycloak	required
realm_spec	KeycloakRealmSpec	Desired realm specification	required

Returns:

Type	Description
bool	True if configuration matches, False otherwise

delete_keycloak_realm async

delete_keycloak_realm(
    spec,
    name,
    namespace,
    status,
    patch,
    memo,
    retry,
    **kwargs
)

Handle KeycloakRealm deletion.

This handler performs comprehensive cleanup of the realm from Keycloak and any associated Kubernetes resources. Finalizer management is handled automatically by Kopf via settings.persistence.finalizer.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	KeycloakRealm resource specification	required
name	str	Name of the KeycloakRealm resource	required
namespace	str	Namespace where the resource exists	required
status	dict[str, Any]	Current status of the resource	required
patch	Patch	Kopf patch object for modifying the resource	required
retry	int	Kopf retry count (starts from 0)	required

ensure_keycloak_realm async

ensure_keycloak_realm(
    spec, name, namespace, status, patch, memo, **kwargs
)

Ensure KeycloakRealm exists in the target Keycloak instance.

This handler creates and configures realms in Keycloak instances. Realms can be created in Keycloak instances across namespaces, subject to RBAC permissions.

Note: Deletion is handled by the @kopf.on.delete handler (delete_keycloak_realm). Do not add deletion logic here to avoid race conditions with the delete handler.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	KeycloakRealm resource specification	required
name	str	Name of the KeycloakRealm resource	required
namespace	str	Namespace where the KeycloakRealm resource exists	required
status	dict[str, Any]	Current status of the resource	required
patch	Patch	Kopf patch object for modifying the resource	required

Returns:

Type	Description
None	Dictionary with status information for the resource

monitor_realm_health async

monitor_realm_health(
    stopped,
    spec,
    name,
    namespace,
    status,
    patch,
    meta,
    memo,
    **_kwargs
)

Periodic health check daemon for KeycloakRealms.

This daemon verifies that realms still exist in Keycloak and that their configuration matches the desired state.

Uses a daemon instead of a timer to work around Kopf 1.40.x not supporting callable initial_delay (lambda). The daemon sleeps a random jitter on startup to spread reconciliation load, then loops at TIMER_INTERVAL_REALM between iterations.

The interval is configurable via TIMER_INTERVAL_REALM environment variable. Default: 600 seconds (10 minutes).

Parameters:

Name	Type	Description	Default
stopped	DaemonStopped	Kopf daemon stopped flag for graceful shutdown	required
spec	dict[str, Any]	KeycloakRealm resource specification	required
name	str	Name of the KeycloakRealm resource	required
namespace	str	Namespace where the resource exists	required
status	dict[str, Any]	Current status of the resource	required
patch	Patch	Kopf patch object for modifying the resource	required
meta	dict[str, Any]	Resource metadata	required
memo	Memo	Kopf memo for accessing shared state like rate_limiter	required

update_keycloak_realm async

update_keycloak_realm(
    old,
    new,
    diff,
    name,
    namespace,
    status,
    patch,
    memo,
    **kwargs
)

Handle updates to KeycloakRealm specifications.

This handler processes changes to realm configurations and applies them to the target Keycloak instance.

Parameters:

Name	Type	Description	Default
old	dict[str, Any]	Previous specification	required
new	dict[str, Any]	New specification	required
diff	Diff	List of changes	required
name	str	Name of the KeycloakRealm resource	required
namespace	str	Namespace where the resource exists	required
status	dict[str, Any]	Current status of the resource	required
patch	Patch	Kopf patch object for modifying the resource	required

Returns:

Type	Description
dict[str, Any] | None	None to avoid Kopf creating status subpaths

KeycloakClient handlers - Manages dynamic client provisioning across namespaces.

This module implements the core feature of dynamic client provisioning that enables GitOps-compatible client management. Key features:

• Cross-namespace client creation: Clients can reference Keycloak instances in different namespaces (subject to RBAC permissions)
• RBAC-based authorization: Uses Kubernetes RBAC instead of Keycloak's built-in security mechanisms
• Secure secret management: Client credentials stored in Kubernetes secrets with proper access controls
• GitOps compatibility: All client configuration is declarative

The handlers support various client types including: - Public clients (SPAs, mobile apps) - Confidential clients (backend services) - Service accounts for machine-to-machine communication

StatusWrapper

StatusWrapper(patch_status)

Wrapper to make kopf patch.status compatible with StatusProtocol.

This wrapper provides both attribute and dict-like access to patch.status, ensuring all updates are written directly to the underlying patch object.

Automatically converts snake_case Python attribute names to camelCase for K8s API.

_to_camel_case staticmethod

_to_camel_case(snake_str)

Convert snake_case to camelCase.

update

update(data)

Update multiple fields. Assumes data keys are already in camelCase.

_calculate_exponential_backoff

_calculate_exponential_backoff(
    retry_count,
    initial_backoff=ROTATION_INITIAL_BACKOFF_SECONDS,
    max_backoff=ROTATION_MAX_BACKOFF_SECONDS,
)

Calculate exponential backoff delay.

Parameters:

Name	Type	Description	Default
retry_count	int	Current retry attempt number (0-based)	required
initial_backoff	float	Initial backoff delay in seconds	ROTATION_INITIAL_BACKOFF_SECONDS
max_backoff	float	Maximum backoff delay in seconds	ROTATION_MAX_BACKOFF_SECONDS

Returns:

Type	Description
float	Backoff delay in seconds

_calculate_seconds_until_rotation

_calculate_seconds_until_rotation(spec, rotated_at, logger)

Calculate the number of seconds until the next rotation is due.

Takes into account both the rotation period and optional rotation time window.

Parameters:

Name	Type	Description	Default
spec	KeycloakClientSpec	Client specification with rotation settings	required
rotated_at	datetime	When the secret was last rotated	required
logger	Logger	Logger for debugging	required

Returns:

Type	Description
float	Seconds until rotation is due (can be 0 if already due)

_parse_duration

_parse_duration(duration_str)

Parse duration string (e.g. '90d', '24h', '10s') into timedelta.

_parse_rotation_timestamp

_parse_rotation_timestamp(timestamp_str)

Parse a rotation timestamp string to a datetime object.

Handles ISO format timestamps and ensures timezone awareness (defaults to UTC).

Parameters:

Name	Type	Description	Default
timestamp_str	str | None	ISO format timestamp string, or None	required

Returns:

Type	Description
datetime | None	Parsed datetime with timezone, or None if input was None/invalid

_perform_client_cleanup async

_perform_client_cleanup(
    name,
    namespace,
    spec,
    status,
    rate_limiter,
    trigger="delete_handler",
)

Perform client cleanup from Keycloak.

This is the core cleanup logic for delete handlers. Finalizer management is handled by Kopf via settings.persistence.finalizer.

Parameters:

Name	Type	Description	Default
name	str	Name of the KeycloakClient resource	required
namespace	str	Namespace where the resource exists	required
spec	dict[str, Any]	KeycloakClient resource specification	required
status	dict[str, Any]	Current status of the resource	required
rate_limiter	Any	Rate limiter for Keycloak API calls	required
trigger	str	What triggered this cleanup (for logging)	'delete_handler'

_run_client_health_check async

_run_client_health_check(
    spec, name, namespace, status, patch, meta, memo
)

Execute one health check iteration for a KeycloakClient.

delete_keycloak_client async

delete_keycloak_client(
    spec,
    name,
    namespace,
    status,
    patch,
    memo,
    retry,
    **kwargs
)

Handle KeycloakClient deletion.

This handler performs comprehensive cleanup of the client from Keycloak and any associated Kubernetes resources. Finalizer management is handled automatically by Kopf via settings.persistence.finalizer.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	KeycloakClient resource specification	required
name	str	Name of the KeycloakClient resource	required
namespace	str	Namespace where the resource exists	required
status	dict[str, Any]	Current status of the resource	required
patch	Patch	Kopf patch object for modifying the resource	required
retry	int	Kopf retry count (starts from 0)	required

ensure_keycloak_client async

ensure_keycloak_client(
    spec, name, namespace, status, patch, memo, **kwargs
)

Ensure KeycloakClient exists in the target Keycloak instance.

This handler implements dynamic client provisioning across namespaces. It can create clients in Keycloak instances located in any namespace, subject to RBAC permissions.

Note: Deletion is handled by the @kopf.on.delete handler (delete_keycloak_client). Do not add deletion logic here to avoid race conditions with the delete handler.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	KeycloakClient resource specification	required
name	str	Name of the KeycloakClient resource	required
namespace	str	Namespace where the KeycloakClient resource exists	required
status	dict[str, Any]	Current status of the resource	required
patch	Patch	Kopf patch object for modifying the resource	required

Returns:

Type	Description
None	Dictionary with status information for the resource

monitor_client_health async

monitor_client_health(
    stopped,
    spec,
    name,
    namespace,
    status,
    patch,
    meta,
    memo,
    **kwargs
)

Periodic health check daemon for KeycloakClients.

This daemon verifies that clients still exist in Keycloak and that their configuration matches the desired state.

Uses a daemon instead of a timer to work around Kopf 1.40.x not supporting callable initial_delay (lambda). The daemon sleeps a random jitter on startup to spread reconciliation load, then loops at TIMER_INTERVAL_CLIENT between iterations.

The interval is configurable via TIMER_INTERVAL_CLIENT environment variable. Default: 300 seconds (5 minutes).

Parameters:

Name	Type	Description	Default
stopped	DaemonStopped	Kopf daemon stopped flag for graceful shutdown	required
spec	dict[str, Any]	KeycloakClient resource specification	required
name	str	Name of the KeycloakClient resource	required
namespace	str	Namespace where the resource exists	required
status	dict[str, Any]	Current status of the resource	required
meta	dict[str, Any]	Resource metadata	required
memo	Memo	Kopf memo for accessing shared state like rate_limiter	required

monitor_client_secrets async

monitor_client_secrets(event, logger, **kwargs)

Monitor client secrets and trigger reconciliation if deleted.

If a managed secret is deleted, we must trigger reconciliation on the parent KeycloakClient to recreate it.

secret_rotation_daemon async

secret_rotation_daemon(
    spec,
    name,
    namespace,
    status,
    meta,
    stopped,
    patch,
    memo,
    logger,
    **kwargs
)

Daemon to handle secret rotation with precise timing.

This daemon: 1. Reads the rotated-at annotation from the secret to determine last rotation 2. Calculates exactly when the next rotation is due 3. Sleeps precisely until that time (using stopped.wait for clean shutdown) 4. Performs the rotation with retry logic 5. On persistent failure, sets status to Degraded (manual intervention needed)

The daemon only runs for confidential clients with secretRotation.enabled=true.

update_keycloak_client async

update_keycloak_client(
    old,
    new,
    diff,
    name,
    namespace,
    status,
    patch,
    memo,
    **kwargs
)

Handle updates to KeycloakClient specifications.

This handler processes changes to client configurations and applies them to the target Keycloak instance.

Parameters:

Name	Type	Description	Default
old	dict[str, Any]	Previous specification	required
new	dict[str, Any]	New specification	required
diff	Diff	List of changes	required
name	str	Name of the KeycloakClient resource	required
namespace	str	Namespace where the resource exists	required
status	dict[str, Any]	Current status of the resource	required
patch	Patch	Kopf patch object for modifying the resource	required

Returns:

Type	Description
dict[str, Any] | None	None to avoid Kopf creating status subpaths

Services

Keycloak instance reconciler for managing core Keycloak deployments.

This module handles the lifecycle of Keycloak instances including deployment, services, persistence, and administrative access.

KeycloakInstanceReconciler

KeycloakInstanceReconciler(
    k8s_client=None,
    keycloak_admin_factory=None,
    rate_limiter=None,
)

Bases: BaseReconciler

Reconciler for Keycloak instance resources.

Manages the complete lifecycle of Keycloak instances including: - Kubernetes deployment and scaling - Service and ingress configuration - Persistent storage setup - Admin user initialization - Production environment validation

Initialize Keycloak instance reconciler.

Parameters:

Name	Type	Description	Default
k8s_client	ApiClient | None	Kubernetes API client	None
keycloak_admin_factory	Any	Factory function for creating Keycloak admin clients	None
rate_limiter	Any	Rate limiter for Keycloak API calls	None

_check_pod_logs_for_build_errors

_check_pod_logs_for_build_errors(
    deployment_name, namespace
)

Check pod logs for known build/configuration mismatch errors.

Returns:

Type	Description
str | None	Error message if a known error is found, None otherwise.

_delete_configuration_resources async

_delete_configuration_resources(name, namespace, core_api)

Delete secrets and configmaps associated with the Keycloak instance.

_delete_dependent_resources async

_delete_dependent_resources(
    keycloak_name, keycloak_namespace
)

Delete all CRD resources that depend on this Keycloak instance.

This implements cascading deletion by finding and deleting: 1. KeycloakClients that reference this Keycloak instance 2. KeycloakRealms that reference this Keycloak instance

Parameters:

Name	Type	Description	Default
keycloak_name	str	Name of the Keycloak instance being deleted	required
keycloak_namespace	str	Namespace of the Keycloak instance	required

_delete_deployment async

_delete_deployment(deployment_name, namespace, apps_api)

Delete deployment and wait for pods to terminate.

_delete_ingress async

_delete_ingress(ingress_name, namespace, networking_api)

Delete ingress resource.

_delete_persistent_storage async

_delete_persistent_storage(name, namespace, core_api)

Delete persistent volume claims associated with the Keycloak instance.

_delete_service async

_delete_service(service_name, namespace, core_api)

Delete service resource.

_record_cnpg_status staticmethod

_record_cnpg_status(namespace, healthy)

Record CNPG cluster status to Prometheus metrics.

_update_capacity_status async

_update_capacity_status(status, spec, namespace)

Update realm capacity status fields.

Parameters:

Name	Type	Description	Default
status	StatusProtocol	Status object to update	required
spec	KeycloakSpec	Keycloak specification	required
namespace	str	Keycloak namespace	required

_update_deployment async

_update_deployment(
    deployment_name, namespace, deployment_changes, apps_api
)

Update deployment with changes.

_update_ingress async

_update_ingress(spec, name, namespace, networking_api)

Update ingress configuration.

_validate_database_connectivity async

_validate_database_connectivity(spec, name, namespace)

Validate database connectivity using the new database connection manager.

Parameters:

Name	Type	Description	Default
spec	KeycloakSpec	Keycloak specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

Returns:

Type	Description
dict[str, Any]	Dictionary with resolved database connection details

Raises:

Type	Description
ExternalServiceError	If database connectivity validation fails

_validate_database_production_readiness async

_validate_database_production_readiness(spec, namespace)

Validate database configuration for production readiness.

Parameters:

Name	Type	Description	Default
spec	KeycloakSpec	Keycloak specification	required
namespace	str	Resource namespace	required

Raises:

Type	Description
DatabaseValidationError	If database configuration is not production-ready

_validate_database_secret async

_validate_database_secret(spec, namespace)

Validate that database credentials secret exists.

Parameters:

Name	Type	Description	Default
spec	KeycloakSpec	Keycloak specification	required
namespace	str	Resource namespace	required

Raises:

Type	Description
ExternalServiceError	If database secret is missing or invalid

_validate_security_requirements async

_validate_security_requirements(spec, namespace)

Validate security requirements for production deployments.

Parameters:

Name	Type	Description	Default
spec	KeycloakSpec	Keycloak specification	required
namespace	str	Resource namespace	required

Raises:

Type	Description
ValidationError	If security requirements are not met

_validate_spec

_validate_spec(spec)

Validate and parse Keycloak specification.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	Raw specification dictionary	required

Returns:

Type	Description
KeycloakSpec	Validated KeycloakSpec object

Raises:

Type	Description
ValidationError	If specification is invalid

_wait_for_rollout async

_wait_for_rollout(deployment_name, namespace, apps_api)

Wait for deployment rollout to complete.

cleanup_resources async

cleanup_resources(name, namespace, spec)

Clean up all resources associated with a Keycloak instance.

This method performs comprehensive cleanup in the proper order to prevent data loss and ensure all associated resources are properly removed.

Implements cascading deletion: 1. Delete dependent KeycloakClients 2. Delete dependent KeycloakRealms 3. Delete Kubernetes resources (deployments, services, etc.)

Parameters:

Name	Type	Description	Default
name	str	Name of the Keycloak instance	required
namespace	str	Namespace containing the resources	required
spec	dict[str, Any]	Keycloak specification for understanding deletion requirements	required

Raises:

Type	Description
TemporaryError	If cleanup fails but should be retried

do_reconcile async

do_reconcile(spec, name, namespace, status, **kwargs)

Reconcile Keycloak instance to desired state.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	Keycloak resource specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required
status	StatusProtocol	Resource status object	required
**kwargs	Any	Additional handler arguments	{}

Returns:

Type	Description
dict[str, Any]	Status dictionary for the resource

do_update async

do_update(
    old_spec,
    new_spec,
    diff,
    name,
    namespace,
    status,
    **kwargs
)

Handle updates to Keycloak instance specifications.

Parameters:

Name	Type	Description	Default
old_spec	dict[str, Any]	Previous specification	required
new_spec	dict[str, Any]	New specification	required
diff	Any	List of changes between old and new	required
name	str	Name of the Keycloak resource	required
namespace	str	Namespace where the resource exists	required
status	StatusProtocol	Current status of the resource	required
**kwargs	Any	Additional handler arguments	{}

Returns:

Type	Description
dict[str, Any] | None	Dictionary with updated status information, or None if no changes needed

ensure_admin_access async

ensure_admin_access(spec, name, namespace)

Ensure admin user is configured and accessible.

Parameters:

Name	Type	Description	Default
spec	KeycloakSpec	Keycloak specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

ensure_deployment async

ensure_deployment(
    spec, name, namespace, db_connection_info=None
)

Ensure Keycloak deployment exists and is up to date.

Parameters:

Name	Type	Description	Default
spec	KeycloakSpec	Keycloak specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required
db_connection_info	dict[str, Any] | None	Optional resolved database connection details	None

ensure_discovery_service async

ensure_discovery_service(name, namespace)

Ensure headless discovery service exists for JGroups clustering.

This headless service enables Keycloak replicas to discover each other via DNS_PING for proper session replication and cache synchronization.

Parameters:

Name	Type	Description	Default
name	str	Resource name	required
namespace	str	Resource namespace	required

ensure_ingress async

ensure_ingress(spec, name, namespace)

Ensure ingress is configured for external access.

Parameters:

Name	Type	Description	Default
spec	KeycloakSpec	Keycloak specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

ensure_service async

ensure_service(spec, name, namespace)

Ensure Keycloak service exists and is properly configured.

Parameters:

Name	Type	Description	Default
spec	KeycloakSpec	Keycloak specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

validate_production_settings async

validate_production_settings(spec, name, namespace)

Validate configuration for production readiness.

Parameters:

Name	Type	Description	Default
spec	KeycloakSpec	Keycloak specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

Returns:

Type	Description
dict[str, Any] | None	Dictionary with resolved database connection details, or None if database not configured

Raises:

Type	Description
DatabaseValidationError	If using H2 database inappropriately
ValidationError	If other production requirements not met

wait_for_deployment_ready async

wait_for_deployment_ready(
    name, namespace, max_wait_time=300
)

Wait for deployment to be ready.

Parameters:

Name	Type	Description	Default
name	str	Resource name	required
namespace	str	Resource namespace	required
max_wait_time	int	Maximum wait time in seconds	300

Returns:

Type	Description
tuple[bool, str | None]	Tuple of (ready: bool, error_message: str | None)

Keycloak realm reconciler for managing realm configuration and features.

This module handles the lifecycle of Keycloak realms including themes, authentication flows, identity providers, and user federation.

KeycloakRealmReconciler

KeycloakRealmReconciler(
    k8s_client=None,
    keycloak_admin_factory=None,
    rate_limiter=None,
)

Bases: BaseReconciler

Reconciler for Keycloak realm resources.

Manages the complete configuration of Keycloak realms including: - Basic realm creation and settings - Theme and branding configuration - Authentication flows and security - Identity provider integration - User federation setup - Cross-namespace RBAC validation

Initialize Keycloak realm reconciler.

Parameters:

Name	Type	Description	Default
k8s_client	ApiClient | None	Kubernetes API client	None
keycloak_admin_factory	Any	Factory function for creating Keycloak admin clients	None
rate_limiter	Any	Rate limiter for Keycloak API calls	None

_add_flow_executions async

_add_flow_executions(
    admin_client,
    realm_name,
    flow_alias,
    flow_config,
    namespace,
)

Add executions to a newly created flow.

Parameters:

Name	Type	Description	Default
admin_client		Keycloak admin client	required
realm_name	str	Name of the realm	required
flow_alias	str	Alias of the flow	required
flow_config		Flow configuration with executions	required
namespace	str	Origin namespace for rate limiting	required

_build_federation_component async

_build_federation_component(
    federation_config, realm_id, namespace
)

Build a ComponentRepresentation from KeycloakUserFederation config.

Handles secret injection for bind credentials and keytabs.

Parameters:

Name	Type	Description	Default
federation_config	KeycloakUserFederation	The user federation configuration	required
realm_id	str	The realm ID (UUID), not realm name - required for parentId	required
namespace	str	Kubernetes namespace for secret lookups	required

_build_group_map

_build_group_map(groups, group_map, parent_path='')

Build a map of group paths to group info (including subgroups).

_check_realm_capacity async

_check_realm_capacity(
    keycloak_namespace, keycloak_name, realm_name
)

Check if Keycloak operator has capacity for new realms.

Parameters:

Name	Type	Description	Default
keycloak_namespace	str	Namespace containing the Keycloak instance	required
keycloak_name	str	Name of the Keycloak instance	required
realm_name	str	Name of the realm being created	required

Raises:

Type	Description
PermanentError	If capacity is exhausted and new realms are not allowed

_collect_group_paths

_collect_group_paths(groups, paths, parent_path='')

Collect all group paths from the spec (including subgroups).

_configure_authenticator_configs async

_configure_authenticator_configs(
    admin_client,
    realm_name,
    flow_alias,
    flow_config,
    namespace,
)

Configure authenticator configurations for executions in a flow.

Parameters:

Name	Type	Description	Default
admin_client		Keycloak admin client	required
realm_name	str	Name of the realm	required
flow_alias	str	Alias of the flow	required
flow_config		Flow configuration with authenticator configs	required
namespace	str	Origin namespace for rate limiting	required

_configure_composite_roles async

_configure_composite_roles(
    admin_client,
    realm_name,
    parent_role_name,
    desired_child_names,
    namespace,
)

Configure composite role memberships.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
parent_role_name	str	Name of the parent composite role	required
desired_child_names	list[str]	List of child role names to include	required
namespace	str	Origin namespace for rate limiting	required

_configure_federation_mappers async

_configure_federation_mappers(
    admin_client,
    realm_name,
    provider_id,
    federation_config,
    namespace,
)

Configure mappers for a user federation provider.

_configure_group_client_roles async

_configure_group_client_roles(
    admin_client,
    realm_name,
    group_id,
    client_roles,
    namespace,
)

Configure client role assignments for a group.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
group_id	str	ID of the group	required
client_roles	dict[str, list[str]]	Dict mapping client IDs to list of role names	required
namespace	str	Origin namespace for rate limiting	required

_configure_group_realm_roles async

_configure_group_realm_roles(
    admin_client,
    realm_name,
    group_id,
    desired_role_names,
    namespace,
)

Configure realm role assignments for a group.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
group_id	str	ID of the group	required
desired_role_names	list[str]	List of realm role names to assign	required
namespace	str	Origin namespace for rate limiting	required

_configure_group_recursive async

_configure_group_recursive(
    admin_client,
    realm_name,
    group_config,
    parent_id,
    namespace,
    parent_path="",
)

Configure a group and its subgroups recursively.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
group_config	Any	Group configuration	required
parent_id	str | None	ID of parent group (None for top-level)	required
namespace	str	Origin namespace for rate limiting	required
parent_path	str	Path of parent group	''

Returns:

Type	Description
str | None	Group ID if created/updated successfully, None otherwise

_configure_identity_provider_mappers async

_configure_identity_provider_mappers(
    admin_client,
    realm_name,
    idp_alias,
    desired_mappers,
    namespace,
)

Configure mappers for an identity provider with full lifecycle management.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
idp_alias	str	Identity provider alias	required
desired_mappers	list[Any]	List of desired mapper configurations	required
namespace	str	Origin namespace for rate limiting	required

_delete_realm_k8s_resources async

_delete_realm_k8s_resources(name, namespace, realm_name)

Delete Kubernetes resources associated with the realm.

_fetch_secret_value async

_fetch_secret_value(namespace, secret_name, secret_key)

Fetch secret value from Kubernetes secret with RBAC validation.

This method enforces namespace access control and secret labeling requirements. The secret must: 1. Be accessible via RoleBinding granting operator access to the namespace 2. Have the label: vriesdemichael.github.io/keycloak-allow-operator-read=true

Parameters:

Name	Type	Description	Default
namespace	str	Namespace containing the secret	required
secret_name	str	Name of the secret	required
secret_key	str	Key in secret data	required

Returns:

Type	Description
str	Decoded secret value string

Raises:

Type	Description
ValidationError	If RBAC validation fails, secret not found, or key missing

_has_flow_bindings

_has_flow_bindings(spec)

Check if the realm spec has any flow bindings configured.

_reconcile_organization_idp_links async

_reconcile_organization_idp_links(
    admin_client,
    realm_name,
    org_id,
    org_name,
    desired_idps,
    namespace,
)

Reconcile identity provider links for an organization.

This method ensures the organization has exactly the IdPs specified in the spec: - Links IdPs that are in the spec but not currently linked - Unlinks IdPs that are currently linked but not in the spec

Parameters:

Name	Type	Description	Default
admin_client	KeycloakAdminClient	Keycloak admin client	required
realm_name	str	Name of the realm	required
org_id	str	UUID of the organization	required
org_name	str	Name of the organization (for logging)	required
desired_idps	list[OrganizationIdentityProvider]	List of desired identity provider links	required
namespace	str	Namespace for rate limiting	required

_sync_client_scope_protocol_mappers async

_sync_client_scope_protocol_mappers(
    admin_client,
    realm_name,
    scope_id,
    desired_mappers,
    namespace,
)

Synchronize protocol mappers for a client scope.

Parameters:

Name	Type	Description	Default
admin_client		Keycloak admin client	required
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope	required
desired_mappers	list	List of desired mapper configurations	required
namespace	str	Origin namespace for rate limiting	required

_sync_flow_executions async

_sync_flow_executions(
    admin_client,
    realm_name,
    flow_alias,
    flow_config,
    namespace,
)

Synchronize executions for an existing flow.

This updates execution requirements to match the desired state. Note: Adding/removing executions from existing flows is complex and not fully implemented - we only update requirements for now.

Parameters:

Name	Type	Description	Default
admin_client		Keycloak admin client	required
realm_name	str	Name of the realm	required
flow_alias	str	Alias of the flow	required
flow_config		Flow configuration with desired executions	required
namespace	str	Origin namespace for rate limiting	required

_validate_spec

_validate_spec(spec)

Validate and parse Keycloak realm specification.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	Raw specification dictionary	required

Returns:

Type	Description
KeycloakRealmSpec	Validated KeycloakRealmSpec object

Raises:

Type	Description
ValidationError	If specification is invalid

_validate_version_compatibility async

_validate_version_compatibility(
    realm_spec, namespace, status, generation
)

Validate realm spec against target Keycloak version.

This method: 1. Gets the admin client to determine Keycloak version 2. Validates the spec against version capabilities 3. Adds version compatibility conditions to CR status

Parameters:

Name	Type	Description	Default
realm_spec	KeycloakRealmSpec	Validated realm specification	required
namespace	str	Resource namespace	required
status	StatusProtocol	Resource status object	required
generation	int	Resource generation for observedGeneration	required

Returns:

Type	Description
ValidationResult	ValidationResult with any warnings or errors

apply_flow_bindings async

apply_flow_bindings(spec, name, namespace)

Apply authentication flow bindings to the realm.

This must be called AFTER authentication flows are created, as flow bindings reference flows by alias.

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

check_resource_exists async

check_resource_exists(name, namespace, spec, status)

Check if realm resource actually exists in Keycloak.

Parameters:

Name	Type	Description	Default
name	str	Name of the KeycloakRealm resource	required
namespace	str	Namespace containing the resource	required
spec	dict[str, Any]	Realm specification	required
status	StatusProtocol	Resource status	required

Returns:

Type	Description
bool	True if realm exists in Keycloak, False otherwise

cleanup_resources async

cleanup_resources(name, namespace, spec, status)

Clean up realm from Keycloak and associated Kubernetes resources.

Parameters:

Name	Type	Description	Default
name	str	Name of the KeycloakRealm resource	required
namespace	str	Namespace containing the resource	required
spec	dict[str, Any]	Realm specification	required
status	StatusProtocol	Resource status for tracking cleanup progress	required

Raises:

Type	Description
TemporaryError	If cleanup fails but should be retried

configure_authentication async

configure_authentication(spec, name, namespace)

Configure authentication flows and security settings.

This method handles: - Creating new authentication flows - Copying from built-in flows (when copyFrom is specified) - Adding executions to flows - Configuring execution requirements - Setting up authenticator configurations

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_client_profiles_and_policies async

configure_client_profiles_and_policies(
    spec, name, namespace
)

Configure client profiles and policies for the realm.

Client profiles define sets of executors that enforce client behavior (e.g., PKCE, secure authentication, redirect URI validation). Client policies define conditions for when profiles should be applied.

This is a full replace operation - all realm-level profiles/policies are replaced with the desired state from the CR.

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_client_scopes async

configure_client_scopes(spec, name, namespace)

Configure client scopes with full lifecycle management.

This method: 1. Creates new client scopes 2. Updates existing client scopes 3. Deletes client scopes removed from spec (except built-in scopes) 4. Manages protocol mappers for each client scope

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_default_groups async

configure_default_groups(spec, name, namespace)

Configure default groups for the realm.

Default groups are automatically assigned to new users.

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_groups async

configure_groups(spec, name, namespace)

Configure groups with full lifecycle management.

This method: 1. Creates new groups (including nested subgroups) 2. Updates existing groups 3. Deletes groups removed from spec 4. Manages group role assignments (realm and client roles)

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_identity_providers async

configure_identity_providers(spec, name, namespace)

Configure external identity providers with full lifecycle management.

This method: 1. Creates new identity providers 2. Updates existing identity providers 3. Deletes identity providers removed from spec 4. Manages IDP mappers for each identity provider

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_organizations async

configure_organizations(spec, name, namespace)

Configure organizations for the realm (Keycloak 26+).

Organizations provide multi-tenancy support, allowing users to be grouped by organization with their own domains and identity providers.

This method performs a full reconciliation: - Creates organizations that don't exist - Updates organizations that have changed - Deletes organizations not in the spec

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

Note

Organizations require Keycloak 26.0.0 or higher. The admin client methods will return empty results or 404 if the feature is not available.

configure_realm_default_client_scopes async

configure_realm_default_client_scopes(
    spec, name, namespace
)

Configure realm-level default and optional client scopes.

Default scopes are automatically assigned to new clients. Optional scopes are available for clients to request.

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_realm_roles async

configure_realm_roles(spec, name, namespace)

Configure realm-level roles with full lifecycle management.

This method: 1. Creates new realm roles 2. Updates existing realm roles 3. Deletes realm roles removed from spec (except built-in roles) 4. Manages composite role memberships

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_required_actions async

configure_required_actions(spec, name, namespace)

Configure required actions for the realm.

Required actions are actions users must perform, such as: - CONFIGURE_TOTP: Set up two-factor authentication - VERIFY_EMAIL: Verify email address - UPDATE_PASSWORD: Change password - UPDATE_PROFILE: Update user profile

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_themes async

configure_themes(spec, name, namespace)

Configure themes and branding for the realm.

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_user_federation async

configure_user_federation(spec, name, namespace)

Configure user federation providers with full CRUD lifecycle.

This method: 1. Fetches existing providers from Keycloak 2. Creates new providers defined in spec 3. Updates existing providers with changed config 4. Deletes providers removed from spec 5. Manages mappers for each provider 6. Injects secrets for bind credentials and keytabs

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

Returns:

Type	Description
list[dict[str, Any]]	List of federation status dicts for status update

do_reconcile async

do_reconcile(spec, name, namespace, status, **kwargs)

Reconcile Keycloak realm to desired state.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	Keycloak realm resource specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required
status	StatusProtocol	Resource status object	required
**kwargs	Any	Additional handler arguments	{}

Returns:

Type	Description
dict[str, Any]	Status dictionary for the resource

do_update async

do_update(
    old_spec,
    new_spec,
    diff,
    name,
    namespace,
    status,
    **kwargs
)

Handle updates to Keycloak realm specifications.

Parameters:

Name	Type	Description	Default
old_spec	dict[str, Any]	Previous specification dictionary	required
new_spec	dict[str, Any]	New specification dictionary	required
diff	Any	List of changes between old and new specs	required
name	str	Resource name	required
namespace	str	Resource namespace	required
status	StatusProtocol	Resource status object	required
**kwargs	Any	Additional handler arguments	{}

Returns:

Type	Description
dict[str, Any] | None	Updated status dictionary or None if no changes needed

ensure_realm_exists async

ensure_realm_exists(spec, name, namespace, **kwargs)

Ensure the basic realm exists in Keycloak with ownership tracking.

This method implements ownership tracking to prevent multiple CRs from managing the same realm and to handle orphaned realms properly.

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required
**kwargs	Any	Additional handler arguments (uid, meta, etc.)	{}

validate_cross_namespace_access async

validate_cross_namespace_access(spec, namespace)

Validate RBAC permissions for cross-namespace operations.

Parameters:

Name	Type	Description	Default
spec	KeycloakRealmSpec	Keycloak realm specification	required
namespace	str	Current namespace	required

Raises:

Type	Description
RBACError	If insufficient permissions for cross-namespace access

Keycloak client reconciler for managing OAuth2/OIDC client configuration.

This module handles the lifecycle of Keycloak clients including client creation, credential management, and OAuth2 configuration.

KeycloakClientReconciler

KeycloakClientReconciler(
    k8s_client=None,
    keycloak_admin_factory=None,
    rate_limiter=None,
)

Bases: BaseReconciler

Reconciler for Keycloak client resources.

Manages the complete lifecycle of OAuth2/OIDC clients including: - Client creation and basic configuration - OAuth2/OIDC parameter setup - Credential generation and rotation - Protocol mapper configuration - Client role management - Cross-namespace RBAC validation

Initialize Keycloak client reconciler.

Parameters:

Name	Type	Description	Default
k8s_client	ApiClient | None	Kubernetes API client	None
keycloak_admin_factory	Any	Factory function for creating Keycloak admin clients	None
rate_limiter	Any	Rate limiter for Keycloak API calls	None

_client_role_needs_update

_client_role_needs_update(existing, desired)

Check if a client role needs to be updated.

Parameters:

Name	Type	Description	Default
existing	dict[str, Any]	Existing client role from Keycloak	required
desired	dict[str, Any]	Desired client role configuration	required

Returns:

Type	Description
bool	True if update is needed, False otherwise

_delete_client_k8s_resources async

_delete_client_k8s_resources(name, namespace, client_spec)

Delete Kubernetes resources associated with the client.

_get_keycloak_instance_from_realm

_get_keycloak_instance_from_realm(
    realm_resource_name, realm_namespace
)

Get Keycloak instance name and namespace from a realm's status.

DEPRECATED: Use _get_realm_info instead for new code.

Parameters:

Name	Type	Description	Default
realm_resource_name	str	Name of the realm resource	required
realm_namespace	str	Namespace of the realm resource	required

Returns:

Type	Description
tuple[str, str]	Tuple of (keycloak_namespace, keycloak_name)

_get_realm_info

_get_realm_info(realm_resource_name, realm_namespace)

Get realm information including actual realm name and Keycloak instance.

Parameters:

Name	Type	Description	Default
realm_resource_name	str	Name of the realm Kubernetes resource	required
realm_namespace	str	Namespace of the realm resource	required

Returns:

Type	Description
tuple[str, str, str, str]	Tuple of (actual_realm_name, keycloak_namespace, keycloak_name, realm_resource_name)

Raises:

Type	Description
ApiException	If realm resource cannot be retrieved
PermanentError	If realm spec is invalid

_parse_duration

_parse_duration(duration_str)

Parse duration string (e.g. '90d', '24h', '10s') into timedelta.

_protocol_mapper_needs_update

_protocol_mapper_needs_update(existing, desired)

Check if a protocol mapper needs to be updated.

Parameters:

Name	Type	Description	Default
existing	dict[str, Any]	Existing protocol mapper from Keycloak	required
desired	dict[str, Any]	Desired protocol mapper configuration	required

Returns:

Type	Description
bool	True if update is needed, False otherwise

_reconcile_authorization_permissions async

_reconcile_authorization_permissions(
    admin_client,
    realm_name,
    client_uuid,
    permissions,
    namespace,
    client_id,
)

Reconcile authorization permissions to match desired state.

Permissions tie policies to resources/scopes. They must be reconciled AFTER policies since they reference policies by name.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
client_uuid	str	Client UUID	required
permissions	Any	AuthorizationPermissions object containing permission types	required
namespace	str	Namespace for rate limiting	required
client_id	str	Client ID for logging	required

_reconcile_authorization_policies async

_reconcile_authorization_policies(
    admin_client,
    realm_name,
    client_uuid,
    policies,
    namespace,
    client_id,
)

Reconcile authorization policies to match desired state.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
client_uuid	str	Client UUID	required
policies	Any	AuthorizationPolicies object containing all policy types	required
namespace	str	Namespace for rate limiting	required
client_id	str	Client ID for logging	required

_reconcile_authorization_resources async

_reconcile_authorization_resources(
    admin_client,
    realm_name,
    client_uuid,
    desired_resources,
    namespace,
    client_id,
)

Reconcile authorization resources to match desired state.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
client_uuid	str	Client UUID	required
desired_resources	list	List of desired AuthorizationResource objects	required
namespace	str	Namespace for rate limiting	required
client_id	str	Client ID for logging	required

_reconcile_authorization_scopes async

_reconcile_authorization_scopes(
    admin_client,
    realm_name,
    client_uuid,
    desired_scopes,
    namespace,
    client_id,
)

Reconcile authorization scopes to match desired state.

Parameters:

Name	Type	Description	Default
admin_client	Any	Keycloak admin client	required
realm_name	str	Name of the realm	required
client_uuid	str	Client UUID	required
desired_scopes	list	List of desired AuthorizationScope objects	required
namespace	str	Namespace for rate limiting	required
client_id	str	Client ID for logging	required

_should_rotate_secret

_should_rotate_secret(spec, secret)

Check if client secret should be rotated based on configuration.

Parameters:

Name	Type	Description	Default
spec	KeycloakClientSpec	Client specification	required
secret	V1Secret	Existing Kubernetes secret	required

Returns:

Type	Description
bool	True if rotation is needed

_validate_namespace_authorization async

_validate_namespace_authorization(
    realm_resource,
    realm_name,
    realm_namespace,
    client_namespace,
    client_id,
)

Validate that client's namespace is authorized via realm's grant list.

Parameters:

Name	Type	Description	Default
realm_resource	dict[str, Any]	The KeycloakRealm custom resource	required
realm_name	str	Name of the realm	required
realm_namespace	str	Namespace of the realm	required
client_namespace	str	Namespace of the client	required
client_id	str	ID of the client	required

Raises:

Type	Description
PermanentError	If namespace is not in grant list

_validate_spec

_validate_spec(spec)

Validate and parse Keycloak client specification.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	Raw specification dictionary	required

Returns:

Type	Description
KeycloakClientSpec	Validated KeycloakClientSpec object

Raises:

Type	Description
ValidationError	If specification is invalid

check_resource_exists async

check_resource_exists(name, namespace, spec, status)

Check if client resource actually exists in Keycloak.

Parameters:

Name	Type	Description	Default
name	str	Name of the KeycloakClient resource	required
namespace	str	Namespace containing the resource	required
spec	dict[str, Any]	Client specification	required
status	StatusProtocol	Resource status	required

Returns:

Type	Description
bool	True if client exists in Keycloak, False otherwise

cleanup_resources async

cleanup_resources(name, namespace, spec, status)

Clean up client from Keycloak and associated Kubernetes resources.

Parameters:

Name	Type	Description	Default
name	str	Name of the KeycloakClient resource	required
namespace	str	Namespace containing the resource	required
spec	dict[str, Any]	Client specification	required
status	StatusProtocol	Resource status for tracking cleanup progress	required

Raises:

Type	Description
TemporaryError	If cleanup fails but should be retried

configure_authorization_settings async

configure_authorization_settings(
    spec, client_uuid, name, namespace
)

Configure fine-grained authorization settings (resources and scopes).

This method manages the authorization services configuration for clients that have authorizationServicesEnabled=true, including: - Resource server settings (policy enforcement mode, decision strategy) - Authorization scopes (read, write, delete, etc.) - Protected resources (APIs, documents, etc.)

Parameters:

Name	Type	Description	Default
spec	KeycloakClientSpec	Keycloak client specification	required
client_uuid	str	Client UUID in Keycloak	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_client_scopes async

configure_client_scopes(spec, client_uuid, name, namespace)

Configure client-level default and optional scope assignments.

Parameters:

Name	Type	Description	Default
spec	KeycloakClientSpec	Keycloak client specification	required
client_uuid	str	Client UUID in Keycloak	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_oauth_settings async

configure_oauth_settings(
    spec, client_uuid, name, namespace
)

Configure OAuth2/OIDC parameters for the client.

Parameters:

Name	Type	Description	Default
spec	KeycloakClientSpec	Keycloak client specification	required
client_uuid	str	Client UUID in Keycloak	required
name	str	Resource name	required
namespace	str	Resource namespace	required

configure_protocol_mappers async

configure_protocol_mappers(
    spec, client_uuid, name, namespace
)

Configure protocol mappers for claims and token customization.

Parameters:

Name	Type	Description	Default
spec	KeycloakClientSpec	Keycloak client specification	required
client_uuid	str	Client UUID in Keycloak	required
name	str	Resource name	required
namespace	str	Resource namespace	required

do_reconcile async

do_reconcile(spec, name, namespace, status, **kwargs)

Reconcile Keycloak client to desired state.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	Keycloak client resource specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required
status	StatusProtocol	Resource status object	required
**kwargs	Any	Additional handler arguments	{}

Returns:

Type	Description
dict[str, Any]	Status dictionary for the resource

do_update async

do_update(
    old_spec,
    new_spec,
    diff,
    name,
    namespace,
    status,
    **kwargs
)

Handle updates to Keycloak client specifications.

Parameters:

Name	Type	Description	Default
old_spec	dict[str, Any]	Previous specification dictionary	required
new_spec	dict[str, Any]	New specification dictionary	required
diff	Any	List of changes between old and new specs	required
name	str	Resource name	required
namespace	str	Resource namespace	required
status	StatusProtocol	Resource status object	required
**kwargs	Any	Additional handler arguments	{}

Returns:

Type	Description
dict[str, Any] | None	Updated status dictionary or None if no changes needed

ensure_client_exists async

ensure_client_exists(spec, name, namespace)

Ensure the OAuth2/OIDC client exists in Keycloak.

Parameters:

Name	Type	Description	Default
spec	KeycloakClientSpec	Keycloak client specification	required
name	str	Resource name	required
namespace	str	Resource namespace	required

Returns:

Type	Description
str	Client UUID from Keycloak

manage_client_credentials async

manage_client_credentials(
    spec, client_uuid, name, namespace, owner_uid=None
)

Generate and manage client credentials (secret).

Parameters:

Name	Type	Description	Default
spec	KeycloakClientSpec	Keycloak client specification	required
client_uuid	str	Client UUID in Keycloak	required
name	str	Resource name	required
namespace	str	Resource namespace	required
owner_uid	str | None	UID of the owning KeycloakClient resource	None

manage_client_roles async

manage_client_roles(spec, client_uuid, name, namespace)

Manage client-specific roles and permissions.

Parameters:

Name	Type	Description	Default
spec	KeycloakClientSpec	Keycloak client specification	required
client_uuid	str	Client UUID in Keycloak	required
name	str	Resource name	required
namespace	str	Resource namespace	required

manage_service_account_roles async

manage_service_account_roles(
    spec, client_uuid, name, namespace
)

Manage role mappings for the client's service account user.

validate_cross_namespace_access async

validate_cross_namespace_access(spec, namespace)

Validate RBAC permissions for cross-namespace operations.

Parameters:

Name	Type	Description	Default
spec	KeycloakClientSpec	Keycloak client specification	required
namespace	str	Current namespace	required

Raises:

Type	Description
RBACError	If insufficient permissions for cross-namespace access

Utilities

Kubernetes utilities for the Keycloak operator.

This module provides helper functions for interacting with the Kubernetes API, including resource management, RBAC validation, and cluster operations.

Key functionality: - Kubernetes client management and configuration - Resource creation and management (deployments, services, secrets) - RBAC permission validation - Cross-namespace resource discovery - Status and health monitoring

check_http_health async

check_http_health(url, timeout=5)

Perform HTTP health check against a URL.

Parameters:

Name	Type	Description	Default
url	str	URL to check	required
timeout	int	Request timeout in seconds	5

Returns:

Type	Description
tuple[bool, str | None]	Tuple of (is_healthy, error_message)

check_rbac_permissions

check_rbac_permissions(
    namespace,
    target_namespace,
    resource="keycloaks",
    verb="get",
    api_group=None,
)

Check if the current service account has RBAC permissions for cross-namespace access.

This function performs a Kubernetes SubjectAccessReview to validate that the operator has the necessary permissions to access resources in other namespaces.

Parameters:

Name	Type	Description	Default
namespace	str	Source namespace (where the request originates)	required
target_namespace	str	Target namespace to access	required
resource	str	Kubernetes resource type to check	'keycloaks'
verb	str	Action to perform (get, create, update, delete, etc.)	'get'
api_group	str | None	API group for the resource (None for auto-detection)	None

Returns:

Type	Description
bool	True if permission is granted, False otherwise

create_admin_secret

create_admin_secret(
    name, namespace, username="admin", password=None
)

Create a secret containing Keycloak admin credentials.

Parameters:

Name	Type	Description	Default
name	str	Name of the Keycloak instance	required
namespace	str	Target namespace	required
username	str	Admin username	'admin'
password	str | None	Admin password (generated if not provided)	None

Returns:

Type	Description
V1Secret	Created Secret object

Creates a secret with admin credentials, generating secure passwords when needed. Sets proper labels and ownership for the secret.

create_client_secret

create_client_secret(
    secret_name,
    namespace,
    client_id,
    client_secret,
    keycloak_url,
    realm,
    update_existing=False,
    labels=None,
    annotations=None,
    owner_uid=None,
    owner_name=None,
)

Create or update a Kubernetes secret containing client credentials.

Parameters:

Name	Type	Description	Default
secret_name	str	Name of the secret to create	required
namespace	str	Target namespace	required
client_id	str	Keycloak client ID	required
client_secret	str | None	Client secret (None for public clients)	required
keycloak_url	str	Keycloak server URL	required
realm	str	Realm name	required
update_existing	bool	Whether to update if secret already exists	False
labels	dict[str, str] | None	Optional labels to add to the secret	None
annotations	dict[str, str] | None	Optional annotations to add to the secret	None
owner_uid	str | None	Optional UID of the owning resource for GC	None
owner_name	str | None	Optional name of the owning resource for GC	None

Returns:

Type	Description
V1Secret	Created or updated Secret object

Creates a Kubernetes secret containing client credentials and connection info. Handles both creation and updates with proper encoding and metadata.

create_keycloak_deployment

create_keycloak_deployment(
    name,
    namespace,
    spec,
    k8s_client,
    db_connection_info=None,
)

Create Kubernetes Deployment for a Keycloak instance.

Args:
    name: Name of the Keycloak resource
    namespace: Target namespace
    spec: Keycloak specification
    k8s_client: Kubernetes API client
    db_connection_info: Optional resolved database connection details (for CNPG)

Returns:
    Created Deployment object

This function creates a complete Keycloak deployment with proper configuration.

create_keycloak_discovery_service

create_keycloak_discovery_service(
    name, namespace, k8s_client
)

Create headless Kubernetes Service for JGroups peer discovery.

This headless service (clusterIP: None) enables Keycloak clustering by creating DNS A-records for each pod IP, allowing JGroups DNS_PING discovery.

Parameters:

Name	Type	Description	Default
name	str	Name of the Keycloak resource	required
namespace	str	Target namespace	required
k8s_client	ApiClient	Kubernetes API client	required

Returns:

Type	Description
V1Service	Created headless Service object

The service exposes port 7800 for JGroups TCP communication between Keycloak replicas. Combined with the KC_CACHE_STACK=kubernetes environment variable and JAVA_OPTS_APPEND DNS query, this enables automatic cluster formation.

create_keycloak_ingress

create_keycloak_ingress(name, namespace, spec, k8s_client)

Create Kubernetes Ingress for a Keycloak instance.

Parameters:

Name	Type	Description	Default
name	str	Name of the Keycloak resource	required
namespace	str	Target namespace	required
spec	KeycloakSpec	Keycloak specification	required
k8s_client	ApiClient	Kubernetes API client	required

Returns:

Type	Description
V1Ingress	Created Ingress object

create_keycloak_service

create_keycloak_service(name, namespace, spec, k8s_client)

Create Kubernetes Service for a Keycloak instance.

Args:
    name: Name of the Keycloak resource
    namespace: Target namespace
    spec: Keycloak specification
    k8s_client: Kubernetes API client

Returns:
    Created Service object

This function creates a Kubernetes service with proper port configuration and selectors.

create_persistent_volume_claim

create_persistent_volume_claim(
    name, namespace, size="10Gi", storage_class=None
)

Create a PersistentVolumeClaim for Keycloak data storage.

Parameters:

Name	Type	Description	Default
name	str	Name of the Keycloak instance	required
namespace	str	Target namespace	required
size	str	Storage size (e.g., "10Gi")	'10Gi'
storage_class	str | None	Storage class name (optional)	None

Returns:

Type	Description
V1PersistentVolumeClaim	Created PersistentVolumeClaim object

find_keycloak_instances

find_keycloak_instances(namespace=None)

Find Keycloak instances across namespaces.

Parameters:

Name	Type	Description	Default
namespace	str | None	Specific namespace to search, or None for cluster-wide	None

Returns:

Type	Description
list[dict[str, Any]]	List of Keycloak instance dictionaries

Searches for Keycloak custom resources and returns instances with status information. Handles API errors gracefully and supports both namespace-specific and cluster-wide searches.

get_admin_credentials

get_admin_credentials(name, namespace)

Get admin credentials for a Keycloak instance.

Parameters:

Name	Type	Description	Default
name	str	Name of the Keycloak instance	required
namespace	str	Namespace where the instance is deployed	required

Returns:

Type	Description
tuple[str, str]	Tuple of (username, password)

Raises:

Type	Description
Exception	If credentials cannot be retrieved

get_current_service_account_info

get_current_service_account_info()

Get information about the current service account being used by the operator.

Returns:

Type	Description
dict[str, str]	Dictionary with service account name and namespace

get_deployment_pods

get_deployment_pods(deployment_name, namespace, k8s_client)

Get all pods belonging to a deployment.

Parameters:

Name	Type	Description	Default
deployment_name	str	Name of the deployment	required
namespace	str	Namespace	required
k8s_client	ApiClient	Kubernetes API client	required

Returns:

Type	Description
list[Any]	List of V1Pod objects

get_kubernetes_client

get_kubernetes_client()

Get configured Kubernetes API client.

This function handles both in-cluster and local development configurations.

Returns:
    Configured Kubernetes API client

This function handles both in-cluster and local development configurations.

get_pod_logs

get_pod_logs(name, namespace, k8s_client, tail_lines=50)

Get logs from a pod.

Parameters:

Name	Type	Description	Default
name	str	Name of the pod	required
namespace	str	Namespace of the pod	required
k8s_client	ApiClient	Kubernetes API client	required
tail_lines	int	Number of lines to retrieve	50

Returns:

Type	Description
str	String containing logs

get_pod_resource_usage

get_pod_resource_usage(name, namespace, k8s_client)

Get resource usage metrics for Keycloak pods.

Parameters:

Name	Type	Description	Default
name	str	Name of the Keycloak resource	required
namespace	str	Target namespace	required
k8s_client	ApiClient	Kubernetes API client	required

Returns:

Type	Description
dict[str, Any]	Dictionary with resource usage information

set_owner_reference

set_owner_reference(
    resource,
    owner_name,
    owner_uid,
    owner_kind="Keycloak",
    api_version="vriesdemichael.github.io/v1",
)

Set owner reference for garbage collection.

Parameters:

Name	Type	Description	Default
resource	Any	Kubernetes resource to set owner reference on	required
owner_name	str	Name of the owner resource	required
owner_uid	str	UID of the owner resource	required
owner_kind	str	Kind of the owner resource	'Keycloak'
api_version	str	API version of the owner resource	'vriesdemichael.github.io/v1'

validate_keycloak_reference

validate_keycloak_reference(keycloak_name, namespace)

Validate that a Keycloak instance reference is valid and ready.

Args:
    keycloak_name: Name of the Keycloak instance
    namespace: Namespace where the instance should exist

Returns:
    Keycloak instance details if valid and ready, None otherwise

This function validates Keycloak instance readiness and availability.

Keycloak Admin API client utilities.

This module provides a high-level interface to the Keycloak Admin REST API for managing Keycloak instances, realms, clients, and other resources.

The client handles: - Authentication with Keycloak admin credentials - Session management and token refresh - Error handling and retry logic - Type-safe API interactions - Rate limiting for API protection

KeycloakAdminClient

KeycloakAdminClient(
    server_url,
    username,
    password,
    realm="master",
    client_id="admin-cli",
    verify_ssl=True,
    timeout=60,
    rate_limiter=None,
    version=None,
    keycloak_name=None,
    keycloak_namespace=None,
)

High-level client for Keycloak Admin API operations.

This client provides methods for managing Keycloak resources including
realms, clients, users, and configuration. It handles authentication,
session management, and provides a clean interface for the operator.

This client provides comprehensive Keycloak Admin API operations with authentication, error handling, and retry logic.

Initialize Keycloak Admin client.

Parameters:

Name	Type	Description	Default
server_url	str	Base URL of the Keycloak server	required
username	str	Admin username	required
password	str	Admin password	required
realm	str	Admin realm (default: master)	'master'
client_id	str	Client ID for admin API access	'admin-cli'
verify_ssl	bool	Whether to verify SSL certificates	True
timeout	int	Request timeout in seconds	60
rate_limiter	RateLimiter | None	Optional rate limiter for API call throttling	None
version	str | None	Keycloak version string (e.g., "24.0.5"). If None, auto-detected.	None
keycloak_name	str | None	Name of the Keycloak instance (for metrics labelling)	None
keycloak_namespace	str | None	Namespace of the Keycloak instance (for metrics labelling)	None

__aenter__ async

__aenter__()

Async context manager entry.

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)

Async context manager exit - ensures session cleanup.

_detect_server_version async

_detect_server_version()

Detect Keycloak version and update adapter.

_ensure_authenticated async

_ensure_authenticated()

Ensure we have a valid access token, refreshing if necessary.

_get_client async

_get_client()

Get or create httpx client (lazy initialization with caching).

The client is cached per (server_url, verify_ssl) combination. We also track the event loop ID to detect when a cached client was created in a different event loop (which can happen in tests).

_make_request async

_make_request(
    method,
    endpoint,
    namespace,
    data=None,
    json=None,
    params=None,
)

Make an authenticated request to the Keycloak Admin API.

Parameters:

Name	Type	Description	Default
method	str	HTTP method (GET, POST, PUT, DELETE)	required
endpoint	str	API endpoint (relative to admin base)	required
namespace	str	Origin namespace for rate limiting	required
data	dict[str, Any] | None	Request body data (deprecated, use json parameter)	None
json	dict[str, Any] | None	JSON request body data	None
params	dict[str, Any] | None	Query parameters	None

Returns:

Type	Description
Response	Response object (httpx.Response) with body already buffered

Raises:

Type	Description
KeycloakAdminError	On API errors or rate limit timeouts

_make_validated_request async

_make_validated_request(
    method,
    endpoint,
    namespace,
    request_model=None,
    response_model=None,
    **kwargs
)

Make an authenticated request with automatic Pydantic validation.

This method wraps _make_request to provide automatic validation of request and response data using Pydantic models.

Parameters:

Name	Type	Description	Default
method	str	HTTP method (GET, POST, PUT, DELETE)	required
endpoint	str	API endpoint (relative to admin base)	required
namespace	str	Origin namespace for rate limiting	required
request_model	BaseModel | None	Pydantic model instance to serialize as request body	None
response_model	type[BaseModel] | None	Pydantic model class to validate response data	None
**kwargs	Any	Additional arguments passed to _make_request	{}

Returns:

Type	Description
Any	Validated response model instance if response_model is provided,
Any	otherwise the raw Response object

Raises:

Type	Description
KeycloakAdminError	If the API request fails
ValidationError	If response data doesn't match the expected model

_record_session_metrics

_record_session_metrics()

Record admin session Prometheus metrics after auth or token refresh.

_refresh_token async

_refresh_token()

Refresh the access token using the refresh token.

add_client_default_scope async

add_client_default_scope(
    realm_name, client_uuid, scope_id, namespace="default"
)

Add a client scope to a client's default scopes.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	UUID of the client	required
scope_id	str	ID of the client scope to add	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

add_client_optional_scope async

add_client_optional_scope(
    realm_name, client_uuid, scope_id, namespace="default"
)

Add a client scope to a client's optional scopes.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	UUID of the client	required
scope_id	str	ID of the client scope to add	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

add_default_group async

add_default_group(
    realm_name, group_id, namespace="default"
)

Add a group to the default groups.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_id	str	ID of the group to add as default	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

add_execution_to_flow async

add_execution_to_flow(
    realm_name, flow_alias, provider_id, namespace
)

Add a new authenticator execution to a flow.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
flow_alias	str	Alias of the parent flow	required
provider_id	str	Authenticator provider ID (e.g., 'auth-cookie')	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
str | None	Execution ID if successful, None otherwise

add_realm_default_client_scope async

add_realm_default_client_scope(
    realm_name, scope_id, namespace="default"
)

Add a client scope to realm default client scopes.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope to add	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

add_realm_optional_client_scope async

add_realm_optional_client_scope(
    realm_name, scope_id, namespace="default"
)

Add a client scope to realm optional client scopes.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope to add	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

add_realm_role_composites async

add_realm_role_composites(
    realm_name, role_name, child_roles, namespace="default"
)

Add composite (child) roles to a realm role.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
role_name	str	Name of the parent role	required
child_roles	list[RoleRepresentation] | list[dict[str, Any]]	List of roles to add as composites	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

add_subflow_to_flow async

add_subflow_to_flow(
    realm_name,
    parent_flow_alias,
    subflow_alias,
    provider_id,
    description,
    namespace,
)

Add a new sub-flow execution to a flow.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
parent_flow_alias	str	Alias of the parent flow	required
subflow_alias	str	Alias for the new sub-flow	required
provider_id	str	Flow provider ('basic-flow' or 'client-flow')	required
description	str | None	Description for the sub-flow	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
str | None	Execution ID if successful, None otherwise

assign_client_roles_to_group async

assign_client_roles_to_group(
    realm_name,
    group_id,
    client_uuid,
    roles,
    namespace="default",
)

Assign client roles to a group.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_id	str	ID of the group	required
client_uuid	str	UUID of the client	required
roles	list[RoleRepresentation] | list[dict[str, Any]]	List of roles to assign	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

assign_client_roles_to_user async

assign_client_roles_to_user(
    user_id, client_uuid, role_names, realm_name, namespace
)

Assign client-level roles to a user.

Parameters:

Name	Type	Description	Default
user_id	str	UUID of the user in Keycloak	required
client_uuid	str	UUID of the client in Keycloak	required
role_names	list[str]	List of role names to assign	required
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Raises:

Type	Description
KeycloakAdminError	If assignment fails

Example

await admin_client.assign_client_roles_to_user( user_id="123-456-789", client_uuid="abc-def-ghi", role_names=["admin", "user"], realm_name="my-realm", namespace="default" )

assign_realm_roles_to_group async

assign_realm_roles_to_group(
    realm_name, group_id, roles, namespace="default"
)

Assign realm roles to a group.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_id	str	ID of the group	required
roles	list[RoleRepresentation] | list[dict[str, Any]]	List of roles to assign	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

assign_realm_roles_to_user async

assign_realm_roles_to_user(
    user_id, role_names, realm_name, namespace
)

Assign realm-level roles to a user.

Parameters:

Name	Type	Description	Default
user_id	str	UUID of the user in Keycloak	required
role_names	list[str]	List of role names to assign	required
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Raises:

Type	Description
KeycloakAdminError	If assignment fails

Example

await admin_client.assign_realm_roles_to_user( user_id="123-456-789", role_names=["admin", "user"], realm_name="my-realm", namespace="default" )

authenticate async

authenticate()

Authenticate with Keycloak and obtain access tokens.

Uses username/password grant to obtain access and refresh tokens.

cleanup

cleanup()

Cleanup resources and metrics when client is discarded from cache.

This method is called when the client is removed from the global cache (e.g. on operator shutdown or when replaced by a new instance).

close async

close()

Close method for compatibility with async context manager.

This is a no-op because both the httpx client and the admin client itself are cached globally and reused across reconciliation cycles. Tokens are preserved for reuse; _ensure_authenticated() handles refresh/expiry automatically before every API call.

configure_identity_provider async

configure_identity_provider(
    realm_name, provider_config, namespace
)

Configure identity provider for a realm (create or update).

This method checks if the identity provider already exists: - If it exists, updates it using PUT - If it doesn't exist, creates it using POST

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
provider_config	IdentityProviderRepresentation | dict[str, Any]	Identity provider configuration as IdentityProviderRepresentation or dict	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

from keycloak_operator.models.keycloak_api import IdentityProviderRepresentation

provider = IdentityProviderRepresentation( alias="google", provider_id="google", enabled=True ) success = await admin_client.configure_identity_provider("my-realm", provider, "my-namespace")

configure_identity_provider_mapper async

configure_identity_provider_mapper(
    realm_name, alias, mapper, namespace
)

Configure an identity provider mapper (create or update).

This method finds existing mappers by name and updates them, or creates new ones if they don't exist.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
alias	str	Identity provider alias	required
mapper	IdentityProviderMapperRepresentation | dict[str, Any]	Mapper configuration	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

configure_user_federation async

configure_user_federation(realm_name, federation_config)

Configure user federation for a realm.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
federation_config	ComponentRepresentation | dict[str, Any]	User federation configuration as ComponentRepresentation or dict	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

from keycloak_operator.models.keycloak_api import ComponentRepresentation

federation = ComponentRepresentation( name="ldap", provider_id="ldap" ) success = admin_client.configure_user_federation("my-realm", federation)

copy_authentication_flow async

copy_authentication_flow(
    realm_name, source_flow_alias, new_flow_alias, namespace
)

Copy an existing authentication flow under a new name.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
source_flow_alias	str	Alias of the flow to copy	required
new_flow_alias	str	Alias for the new flow	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

create_authentication_flow async

create_authentication_flow(
    realm_name, flow_config, namespace
)

Create a new authentication flow for a realm.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
flow_config	AuthenticationFlowRepresentation | dict[str, Any]	Authentication flow configuration	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if the flow was created successfully.
bool	False if the flow already exists (409 conflict) or creation failed.
bool	For idempotent operations, callers should check if the flow exists
bool	separately when False is returned, as the flow may already be present.

create_authenticator_config async

create_authenticator_config(
    realm_name, execution_id, config, namespace
)

Create a new authenticator configuration for an execution.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
execution_id	str	ID of the execution to configure	required
config	AuthenticatorConfigRepresentation | dict[str, Any]	Authenticator configuration	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
str | None	Config ID if successful, None otherwise

create_authorization_permission async

create_authorization_permission(
    realm_name,
    client_uuid,
    permission_type,
    permission_data,
    namespace="default",
)

Create an authorization permission.

API: POST /admin/realms/{realm}/clients/{id}/authz/resource-server/permission/{permission_type}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
permission_type	str	Type of permission ('resource' or 'scope')	required
permission_data	dict	Permission configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict | None	Created permission or None

create_authorization_policy async

create_authorization_policy(
    realm_name,
    client_uuid,
    policy_type,
    policy_data,
    namespace="default",
)

Create an authorization policy.

API: POST /admin/realms/{realm}/clients/{id}/authz/resource-server/policy/{type}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
policy_type	str	Policy type (role, user, group, client, time, js, aggregate, regex)	required
policy_data	dict	Policy configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict | None	Created policy representation or None on failure

create_authorization_resource async

create_authorization_resource(
    realm_name, client_uuid, resource, namespace="default"
)

Create an authorization resource.

API: POST /admin/realms/{realm}/clients/{id}/authz/resource-server/resource

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
resource	dict[str, Any]	Resource configuration (name, uris, type, scopes, etc.)	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict[str, Any] | None	Created resource with ID or None on error

create_authorization_scope async

create_authorization_scope(
    realm_name, client_uuid, scope, namespace="default"
)

Create an authorization scope.

API: POST /admin/realms/{realm}/clients/{id}/authz/resource-server/scope

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
scope	dict[str, Any]	Scope configuration (name, displayName, iconUri)	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict[str, Any] | None	Created scope with ID or None on error

create_client async

create_client(client_config, realm_name, namespace)

Create a new client in the specified realm.

Parameters:

Name	Type	Description	Default
client_config	ClientRepresentation | dict[str, Any]	Client configuration as ClientRepresentation or dict	required
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
str | None	Client UUID if successful, None otherwise

Raises:

Type	Description
KeycloakAdminError	If client creation fails

create_client_protocol_mapper async

create_client_protocol_mapper(
    client_uuid, mapper_config, realm_name="master"
)

Create a protocol mapper for a client.

Parameters:

Name	Type	Description	Default
client_uuid	str	UUID of the client in Keycloak	required
mapper_config	ProtocolMapperRepresentation | dict[str, Any]	Protocol mapper configuration as ProtocolMapperRepresentation or dict	required
realm_name	str	Name of the realm	'master'

Returns:

Type	Description
ProtocolMapperRepresentation | None	Created mapper configuration as ProtocolMapperRepresentation or None on error

Example

from keycloak_operator.models.keycloak_api import ProtocolMapperRepresentation

mapper = ProtocolMapperRepresentation( name="email", protocol="openid-connect", protocol_mapper="oidc-usermodel-property-mapper" ) created = admin_client.create_client_protocol_mapper( client_uuid, mapper, "my-realm" )

create_client_role async

create_client_role(
    client_uuid, role_config, realm_name="master"
)

Create a role for a client.

Parameters:

Name	Type	Description	Default
client_uuid	str	UUID of the client in Keycloak	required
role_config	RoleRepresentation | dict[str, Any]	Role configuration as RoleRepresentation or dict	required
realm_name	str	Name of the realm	'master'

Returns:

Type	Description
bool	True if successful, False otherwise

Example

from keycloak_operator.models.keycloak_api import RoleRepresentation

role = RoleRepresentation( name="admin", description="Administrator role" ) success = admin_client.create_client_role(client_uuid, role, "my-realm")

create_client_scope async

create_client_scope(
    realm_name, scope_config, namespace="default"
)

Create a new client scope.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_config	ClientScopeRepresentation | dict[str, Any]	Client scope configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
str | None	ID of the created client scope, or None if creation failed

create_client_scope_protocol_mapper async

create_client_scope_protocol_mapper(
    realm_name, scope_id, mapper_config, namespace="default"
)

Create a protocol mapper in a client scope.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope	required
mapper_config	ProtocolMapperRepresentation | dict[str, Any]	Protocol mapper configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
str | None	ID of the created mapper, or None if creation failed

create_group async

create_group(realm_name, group_config, namespace='default')

Create a top-level group.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_config	GroupRepresentation | dict[str, Any]	Group configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
str | None	Group ID if successful, None otherwise

create_identity_provider async

create_identity_provider(realm_name, idp_config, namespace)

Create a new identity provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
idp_config	dict[str, Any]	Identity provider configuration	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
str | None	ID of the created identity provider or None if failed

create_identity_provider_mapper async

create_identity_provider_mapper(
    realm_name, alias, mapper, namespace
)

Create a mapper for an identity provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
alias	str	Identity provider alias	required
mapper	IdentityProviderMapperRepresentation | dict[str, Any]	Mapper configuration	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

mapper = IdentityProviderMapperRepresentation( name="email-mapper", identity_provider_alias="github", identity_provider_mapper="hardcoded-user-session-attribute-idp-mapper", config={"attribute": "email", "attribute.value": "${CLAIM.email}"} ) success = await admin_client.create_identity_provider_mapper( "my-realm", "github", mapper, "default" )

create_organization async

create_organization(
    realm_name, organization_data, namespace="default"
)

Create an organization.

API: POST /admin/realms/{realm}/organizations

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
organization_data	dict	Organization configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict | None	Created organization or None

create_realm async

create_realm(realm_config, namespace)

Create a new realm in Keycloak.

Parameters:

Name	Type	Description	Default
realm_config	RealmRepresentation | dict[str, Any]	Realm configuration as RealmRepresentation or dict	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
RealmRepresentation	Created realm information as RealmRepresentation

Raises:

Type	Description
KeycloakAdminError	If realm creation fails

create_realm_role async

create_realm_role(
    realm_name, role_config, namespace="default"
)

Create a realm-level role.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
role_config	RoleRepresentation | dict[str, Any]	Role configuration as RoleRepresentation or dict	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

create_subgroup async

create_subgroup(
    realm_name,
    parent_group_id,
    group_config,
    namespace="default",
)

Create a subgroup under a parent group.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
parent_group_id	str	ID of the parent group	required
group_config	GroupRepresentation | dict[str, Any]	Subgroup configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
str | None	Subgroup ID if successful, None otherwise

create_user_federation_mapper async

create_user_federation_mapper(
    realm_name,
    parent_id,
    mapper_config,
    namespace="default",
)

Create a mapper for a user federation provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
parent_id	str	ID of the parent federation provider	required
mapper_config	ComponentRepresentation | dict[str, Any]	Mapper configuration	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
str | None	ID of the created mapper, or None on failure

Example

mapper = ComponentRepresentation( name="email", provider_id="user-attribute-ldap-mapper", provider_type="org.keycloak.storage.ldap.mappers.LDAPStorageMapper", parent_id=federation_id, config={ "ldap.attribute": ["mail"], "user.model.attribute": ["email"], "read.only": ["true"], } ) mapper_id = await admin_client.create_user_federation_mapper( "my-realm", federation_id, mapper )

create_user_federation_provider async

create_user_federation_provider(
    realm_name, federation_config, namespace="default"
)

Create a new user federation provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
federation_config	ComponentRepresentation | dict[str, Any]	Federation provider configuration	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
str | None	ID of the created provider, or None on failure

Example

config = ComponentRepresentation( name="corporate-ldap", provider_id="ldap", provider_type="org.keycloak.storage.UserStorageProvider", config={ "connectionUrl": ["ldap://ldap.example.com:389"], "usersDn": ["ou=People,dc=example,dc=org"], "bindDn": ["cn=admin,dc=example,dc=org"], "bindCredential": ["secret"], "vendor": ["other"], } ) provider_id = await admin_client.create_user_federation_provider( "my-realm", config )

delete_authentication_flow async

delete_authentication_flow(realm_name, flow_id, namespace)

Delete an authentication flow.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
flow_id	str	ID of the flow to delete	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

delete_authorization_permission async

delete_authorization_permission(
    realm_name,
    client_uuid,
    permission_id,
    namespace="default",
)

Delete an authorization permission.

API: DELETE /admin/realms/{realm}/clients/{id}/authz/resource-server/permission/{permission_id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
permission_id	str	UUID of the permission to delete	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if deleted or not found

delete_authorization_policy async

delete_authorization_policy(
    realm_name, client_uuid, policy_id, namespace="default"
)

Delete an authorization policy.

API: DELETE /admin/realms/{realm}/clients/{id}/authz/resource-server/policy/{policy-id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
policy_id	str	UUID of the policy to delete	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if deleted or not found

delete_authorization_resource async

delete_authorization_resource(
    realm_name,
    client_uuid,
    resource_id,
    namespace="default",
)

Delete an authorization resource.

API: DELETE /admin/realms/{realm}/clients/{id}/authz/resource-server/resource/{resource-id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
resource_id	str	UUID of the resource to delete	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if deleted or not found

delete_authorization_scope async

delete_authorization_scope(
    realm_name, client_uuid, scope_id, namespace="default"
)

Delete an authorization scope.

API: DELETE /admin/realms/{realm}/clients/{id}/authz/resource-server/scope/{scope-id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
scope_id	str	UUID of the scope to delete	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if deleted or not found

delete_client async

delete_client(client_id, realm_name, namespace)

Delete a client from the specified realm.

Parameters:

Name	Type	Description	Default
client_id	str	The client ID to delete	required
realm_name	str	Name of the realm (defaults to "master")	required

Returns:

Type	Description
bool	True if successful, False otherwise

delete_client_protocol_mapper async

delete_client_protocol_mapper(
    client_uuid,
    mapper_id,
    realm_name="master",
    namespace="default",
)

Delete a protocol mapper from a client.

Parameters:

Name	Type	Description	Default
client_uuid	str	UUID of the client in Keycloak	required
mapper_id	str	ID of the protocol mapper	required
realm_name	str	Name of the realm	'master'

Returns:

Type	Description
bool	True if successful, False otherwise

delete_client_role async

delete_client_role(
    client_uuid,
    role_name,
    realm_name="master",
    namespace="default",
)

Delete a role from a client.

Parameters:

Name	Type	Description	Default
client_uuid	str	UUID of the client in Keycloak	required
role_name	str	Name of the role to delete	required
realm_name	str	Name of the realm	'master'

Returns:

Type	Description
bool	True if successful, False otherwise

delete_client_scope async

delete_client_scope(
    realm_name, scope_id, namespace="default"
)

Delete a client scope.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope to delete	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if deletion successful, False otherwise

delete_client_scope_protocol_mapper async

delete_client_scope_protocol_mapper(
    realm_name, scope_id, mapper_id, namespace="default"
)

Delete a protocol mapper from a client scope.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope	required
mapper_id	str	ID of the mapper to delete	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if deletion successful, False otherwise

delete_execution async

delete_execution(realm_name, execution_id, namespace)

Delete an execution from a flow.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
execution_id	str	ID of the execution to delete	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

delete_group async

delete_group(realm_name, group_id, namespace='default')

Delete a group.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_id	str	ID of the group to delete	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

delete_identity_provider async

delete_identity_provider(realm_name, alias, namespace)

Delete an identity provider by alias.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
alias	str	Identity provider alias	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful or not found, False on error

Example

success = await admin_client.delete_identity_provider("my-realm", "github", "default")

delete_identity_provider_mapper async

delete_identity_provider_mapper(
    realm_name, alias, mapper_id, namespace
)

Delete a mapper from an identity provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
alias	str	Identity provider alias	required
mapper_id	str	Mapper ID	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful or not found, False on error

delete_organization async

delete_organization(
    realm_name, org_id, namespace="default"
)

Delete an organization.

API: DELETE /admin/realms/{realm}/organizations/{id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
org_id	str	UUID of the organization	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if deleted or not found

delete_realm async

delete_realm(realm_name, namespace)

Delete a realm from Keycloak.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm to delete	required

Returns:

Type	Description
bool	True if successful, False otherwise

delete_realm_role async

delete_realm_role(
    realm_name, role_name, namespace="default"
)

Delete a realm-level role.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
role_name	str	Name of the role to delete	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

delete_user_federation_mapper async

delete_user_federation_mapper(
    realm_name, mapper_id, namespace="default"
)

Delete a user federation mapper.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
mapper_id	str	ID of the mapper to delete	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

delete_user_federation_provider async

delete_user_federation_provider(
    realm_name, provider_id, namespace="default"
)

Delete a user federation provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
provider_id	str	ID of the federation provider to delete	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

export_realm async

export_realm(realm_name, namespace)

Export realm configuration from Keycloak.

Based on OpenAPI spec: GET /admin/realms/{realm} Returns the complete realm representation.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm to export	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
RealmRepresentation | None	Complete realm configuration as RealmRepresentation or None if not found

Example

realm = await client.export_realm("my-realm", "default") if realm: print(f"Realm {realm.realm} has {len(realm.clients or [])} clients")

get_admin_events async

get_admin_events(
    realm_name,
    namespace,
    date_from=None,
    date_to=None,
    operation_types=None,
    resource_types=None,
    resource_path=None,
    max_results=None,
    first_result=None,
)

Get admin events from Keycloak.

Based on OpenAPI spec: GET /admin/realms/{realm}/admin-events

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required
date_from	str | None	Filter events after this date (format: yyyy-MM-dd)	None
date_to	str | None	Filter events before this date (format: yyyy-MM-dd)	None
operation_types	list[str] | None	Filter by operation types (CREATE, UPDATE, DELETE, ACTION)	None
resource_types	list[str] | None	Filter by resource types (REALM, CLIENT, USER, etc.)	None
resource_path	str | None	Filter by resource path (e.g. 'users/123')	None
max_results	int | None	Maximum number of results to return	None
first_result	int | None	Offset for pagination	None

Returns:

Type	Description
list[AdminEventRepresentation]	List of AdminEventRepresentation objects

get_authentication_flow_by_alias async

get_authentication_flow_by_alias(
    realm_name, flow_alias, namespace
)

Get a specific authentication flow by alias.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
flow_alias	str	Alias of the flow	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
AuthenticationFlowRepresentation | None	Authentication flow or None if not found

get_authentication_flows async

get_authentication_flows(realm_name, namespace)

Get all authentication flows for a realm.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
list[AuthenticationFlowRepresentation]	List of authentication flows

get_authenticator_config async

get_authenticator_config(realm_name, config_id, namespace)

Get authenticator configuration by ID.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
config_id	str	ID of the authenticator config	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
AuthenticatorConfigRepresentation | None	Authenticator config or None if not found

get_authorization_permission_by_name async

get_authorization_permission_by_name(
    realm_name,
    client_uuid,
    permission_name,
    namespace="default",
)

Get a specific authorization permission by name.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
permission_name	str	Name of the permission to find	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict | None	Permission dictionary or None if not found

get_authorization_permissions async

get_authorization_permissions(
    realm_name, client_uuid, namespace="default"
)

Get all authorization permissions for a client's resource server.

API: GET /admin/realms/{realm}/clients/{id}/authz/resource-server/permission

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[dict]	List of permission dictionaries

get_authorization_policies async

get_authorization_policies(
    realm_name, client_uuid, namespace="default"
)

Get all authorization policies for a client's resource server.

API: GET /admin/realms/{realm}/clients/{id}/authz/resource-server/policy

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[dict]	List of policy representations

get_authorization_policy_by_name async

get_authorization_policy_by_name(
    realm_name,
    client_uuid,
    policy_name,
    namespace="default",
)

Get a specific authorization policy by name.

API: GET /admin/realms/{realm}/clients/{id}/authz/resource-server/policy?name={name}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
policy_name	str	Name of the policy	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict | None	Policy representation or None if not found

get_authorization_resources async

get_authorization_resources(
    realm_name, client_uuid, namespace="default"
)

Get all authorization resources for a client's resource server.

API: GET /admin/realms/{realm}/clients/{id}/authz/resource-server/resource

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[dict[str, Any]]	List of resource representations

get_authorization_scopes async

get_authorization_scopes(
    realm_name, client_uuid, namespace="default"
)

Get all authorization scopes for a client's resource server.

API: GET /admin/realms/{realm}/clients/{id}/authz/resource-server/scope

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[dict[str, Any]]	List of scope representations

get_client_by_name async

get_client_by_name(client_id, realm_name, namespace)

Get a client by its client ID in the specified realm.

Parameters:

Name	Type	Description	Default
client_id	str	The client ID to search for	required
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
ClientRepresentation | None	Client data as ClientRepresentation if found, None otherwise

Example

client = await admin_client.get_client_by_name("my-client", "my-realm", "default") if client: print(f"Client UUID: {client.id}, Enabled: {client.enabled}")

get_client_default_scopes async

get_client_default_scopes(
    realm_name, client_uuid, namespace="default"
)

Get default client scopes assigned to a specific client.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	UUID of the client	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[ClientScopeRepresentation]	List of default client scopes for this client

get_client_optional_scopes async

get_client_optional_scopes(
    realm_name, client_uuid, namespace="default"
)

Get optional client scopes assigned to a specific client.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	UUID of the client	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[ClientScopeRepresentation]	List of optional client scopes for this client

get_client_policies async

get_client_policies(realm_name, namespace='default')

Get client policies for a realm.

Client policies define conditions for when profiles should be applied. The response includes both realm-level policies and global (built-in) policies.

API: GET /admin/realms/{realm}/client-policies/policies

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict[str, Any] | None	Dict with 'policies' (realm-defined) and 'globalPolicies' (built-in),
dict[str, Any] | None	or None on error

Example

policies_data = await admin_client.get_client_policies("my-realm") if policies_data: for policy in policies_data.get("policies", []): print(f"Policy: {policy['name']}, Enabled: {policy.get('enabled', True)}")

get_client_profiles async

get_client_profiles(realm_name, namespace='default')

Get client profiles for a realm.

Client profiles define sets of executors that enforce client behavior. The response includes both realm-level profiles and global (built-in) profiles.

API: GET /admin/realms/{realm}/client-policies/profiles

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict[str, Any] | None	Dict with 'profiles' (realm-defined) and 'globalProfiles' (built-in),
dict[str, Any] | None	or None on error

Example

profiles_data = await admin_client.get_client_profiles("my-realm") if profiles_data: for profile in profiles_data.get("profiles", []): print(f"Profile: {profile['name']}")

get_client_protocol_mappers async

get_client_protocol_mappers(
    client_uuid, realm_name="master", namespace="default"
)

Get all protocol mappers for a client.

Parameters:

Name	Type	Description	Default
client_uuid	str	UUID of the client in Keycloak	required
realm_name	str	Name of the realm	'master'

Returns:

Type	Description
list[ProtocolMapperRepresentation]	List of protocol mapper configurations as ProtocolMapperRepresentation

Example

mappers = admin_client.get_client_protocol_mappers(client_uuid, "my-realm") for mapper in mappers: print(f"Mapper: {mapper.name}, Protocol: {mapper.protocol}")

get_client_role async

get_client_role(
    client_uuid, role_name, realm_name, namespace
)

Get a client role by name.

Parameters:

Name	Type	Description	Default
client_uuid	str	UUID of the client in Keycloak	required
role_name	str	Name of the role to retrieve	required
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
RoleRepresentation | None	Role representation as RoleRepresentation or None if not found

Example

role = await admin_client.get_client_role(client_uuid, "admin", "my-realm", "default") if role: print(f"Client role: {role.name}, ID: {role.id}")

get_client_roles async

get_client_roles(
    client_uuid, realm_name="master", namespace="default"
)

Get all roles for a client.

Parameters:

Name	Type	Description	Default
client_uuid	str	UUID of the client in Keycloak	required
realm_name	str	Name of the realm	'master'

Returns:

Type	Description
list[RoleRepresentation]	List of client role configurations as RoleRepresentation

Example

roles = admin_client.get_client_roles(client_uuid, "my-realm") for role in roles: print(f"Role: {role.name}, ID: {role.id}")

get_client_scope_by_name async

get_client_scope_by_name(
    realm_name, scope_name, namespace="default"
)

Get a specific client scope by name.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_name	str	Name of the client scope	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
ClientScopeRepresentation | None	ClientScopeRepresentation if found, None otherwise

get_client_scope_protocol_mappers async

get_client_scope_protocol_mappers(
    realm_name, scope_id, namespace="default"
)

Get protocol mappers for a client scope.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[ProtocolMapperRepresentation]	List of protocol mappers

get_client_scopes async

get_client_scopes(realm_name, namespace='default')

Get all client scopes in a realm.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[ClientScopeRepresentation]	List of client scopes as ClientScopeRepresentation

get_client_secret async

get_client_secret(client_id, realm_name, namespace)

Get the client secret for a confidential client.

Parameters:

Name	Type	Description	Default
client_id	str	The client ID	required
realm_name	str	Name of the realm (defaults to "master")	required

Returns:

Type	Description
str | None	Client secret if found, None otherwise

get_client_uuid async

get_client_uuid(client_id, realm_name, namespace)

Get client UUID by client ID in the specified realm.

Parameters:

Name	Type	Description	Default
client_id	str	The client ID to search for	required
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
str | None	Client UUID if found, None otherwise

get_default_groups async

get_default_groups(realm_name, namespace='default')

Get the default groups for a realm.

Default groups are automatically assigned to new users.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[GroupRepresentation]	List of default groups

get_flow_executions async

get_flow_executions(realm_name, flow_alias, namespace)

Get all executions for an authentication flow.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
flow_alias	str	Alias of the flow	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
list[AuthenticationExecutionInfoRepresentation]	List of execution info representations

get_group_by_id async

get_group_by_id(realm_name, group_id, namespace='default')

Get a specific group by its ID.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_id	str	ID of the group	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
GroupRepresentation | None	GroupRepresentation if found, None otherwise

get_group_by_path async

get_group_by_path(realm_name, path, namespace='default')

Get a group by its path (e.g., /parent/child).

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
path	str	Group path (e.g., "/parent/child")	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
GroupRepresentation | None	GroupRepresentation if found, None otherwise

get_group_client_role_mappings async

get_group_client_role_mappings(
    realm_name, group_id, client_uuid, namespace="default"
)

Get client role mappings for a group.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_id	str	ID of the group	required
client_uuid	str	UUID of the client	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[RoleRepresentation]	List of assigned client roles

get_group_realm_role_mappings async

get_group_realm_role_mappings(
    realm_name, group_id, namespace="default"
)

Get realm role mappings for a group.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_id	str	ID of the group	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[RoleRepresentation]	List of assigned realm roles

get_groups async

get_groups(
    realm_name,
    namespace="default",
    brief_representation=False,
)

Get all top-level groups in a realm.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	'default'
brief_representation	bool	If true, returns only basic group info	False

Returns:

Type	Description
list[GroupRepresentation]	List of group configurations

get_identity_provider async

get_identity_provider(realm_name, alias, namespace)

Get an identity provider by alias.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
alias	str	Identity provider alias	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
IdentityProviderRepresentation | None	IdentityProviderRepresentation if found, None otherwise

Example

idp = await admin_client.get_identity_provider("my-realm", "github", "default") if idp: print(f"IdP: {idp.alias}, Enabled: {idp.enabled}")

get_identity_provider_mapper async

get_identity_provider_mapper(
    realm_name, alias, mapper_id, namespace
)

Get a specific mapper by ID.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
alias	str	Identity provider alias	required
mapper_id	str	Mapper ID	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
IdentityProviderMapperRepresentation | None	IdentityProviderMapperRepresentation if found, None otherwise

get_identity_provider_mappers async

get_identity_provider_mappers(realm_name, alias, namespace)

Get all mappers for an identity provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
alias	str	Identity provider alias	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
list[IdentityProviderMapperRepresentation]	List of IdentityProviderMapperRepresentation objects

Example

mappers = await admin_client.get_identity_provider_mappers( "my-realm", "github", "default" ) for mapper in mappers: print(f"Mapper: {mapper.name}, Type: {mapper.identity_provider_mapper}")

get_identity_providers async

get_identity_providers(realm_name, namespace)

Get all identity providers in a realm.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
list[IdentityProviderRepresentation]	List of identity providers

get_latest_admin_event_time async

get_latest_admin_event_time(
    realm_name,
    namespace,
    scope="realm",
    client_uuid=None,
    since_timestamp=None,
)

Get the timestamp of the latest config-changing admin event.

This method is used for drift detection to determine if any configuration changes have been made since the last reconciliation.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required
scope	str	Either "realm" or "client" - determines which resource types to filter	'realm'
client_uuid	str | None	Required when scope="client" - the client UUID to filter events for	None
since_timestamp	int | None	Optional timestamp (Unix ms) to filter events after this time	None

Returns:

Type	Description
int | None	The timestamp (Unix ms) of the latest matching event, or None if no events found

Examples:

Get latest realm config event

ts = await client.get_latest_admin_event_time("my-realm", "default", scope="realm")

Get latest client config event

ts = await client.get_latest_admin_event_time( "my-realm", "default", scope="client", client_uuid="abc-123" )

Check for events since last reconcile

ts = await client.get_latest_admin_event_time( "my-realm", "default", since_timestamp=last_reconcile_time )

get_organization async

get_organization(realm_name, org_id, namespace='default')

Get an organization by ID.

API: GET /admin/realms/{realm}/organizations/{id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
org_id	str	ID of the organization	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict | None	Organization dictionary or None if not found

get_organization_by_name async

get_organization_by_name(
    realm_name, org_name, namespace="default"
)

Get an organization by name.

This method first searches for the organization by name, then fetches the full details by ID to ensure all fields (including attributes) are returned.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
org_name	str	Name of the organization	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict | None	Organization dictionary or None if not found

get_organization_identity_providers async

get_organization_identity_providers(
    realm_name, org_id, namespace="default"
)

Get identity providers linked to an organization.

API: GET /admin/realms/{realm}/organizations/{orgId}/identity-providers

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
org_id	str	UUID of the organization	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[dict]	List of linked identity provider dictionaries

get_organizations async

get_organizations(realm_name, namespace='default')

Get all organizations in a realm.

API: GET /admin/realms/{realm}/organizations

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[dict]	List of organization dictionaries

get_realm async

get_realm(realm_name, namespace)

Get realm configuration from Keycloak.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm to retrieve	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
RealmRepresentation | None	Realm configuration as RealmRepresentation or None if not found

Raises:

Type	Description
KeycloakAdminError	If the request fails (except 404)

get_realm_clients async

get_realm_clients(realm_name, namespace)

Get all clients in a realm.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required

Returns:

Type	Description
list[ClientRepresentation]	List of client configurations as ClientRepresentation

Example

clients = admin_client.get_realm_clients("my-realm") for client in clients: print(f"Client: {client.client_id}, Enabled: {client.enabled}")

get_realm_default_client_scopes async

get_realm_default_client_scopes(
    realm_name, namespace="default"
)

Get realm default client scopes.

These scopes are assigned to all new clients by default.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[ClientScopeRepresentation]	List of default client scopes

get_realm_optional_client_scopes async

get_realm_optional_client_scopes(
    realm_name, namespace="default"
)

Get realm optional client scopes.

These scopes are available for clients to request optionally.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[ClientScopeRepresentation]	List of optional client scopes

get_realm_role async

get_realm_role(role_name, realm_name, namespace)

Get a realm role by name.

Parameters:

Name	Type	Description	Default
role_name	str	Name of the role to retrieve	required
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
RoleRepresentation | None	Role representation as RoleRepresentation or None if not found

Example

role = await admin_client.get_realm_role("admin", "my-realm", "default") if role: print(f"Role: {role.name}, ID: {role.id}")

get_realm_role_by_name async

get_realm_role_by_name(
    realm_name, role_name, namespace="default"
)

Get a specific realm role by name.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
role_name	str	Name of the role	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
RoleRepresentation | None	RoleRepresentation if found, None otherwise

get_realm_role_composites async

get_realm_role_composites(
    realm_name, role_name, namespace="default"
)

Get composite roles (child roles) of a realm role.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
role_name	str	Name of the composite role	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[RoleRepresentation]	List of child role configurations

get_realm_roles async

get_realm_roles(realm_name, namespace='default')

Get all realm-level roles.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
list[RoleRepresentation]	List of realm role configurations as RoleRepresentation

get_realms async

get_realms(namespace, brief_representation=False)

Get all accessible realms from Keycloak.

Based on OpenAPI spec: GET /admin/realms Returns a list of accessible realms filtered by what the caller is allowed to view.

Parameters:

Name	Type	Description	Default
namespace	str	Origin namespace for rate limiting	required
brief_representation	bool	If True, return brief representation of realms	False

Returns:

Type	Description
list[RealmRepresentation] | None	List of realm configurations as RealmRepresentation or None on error

Example

realms = await client.get_realms("default") for realm in realms: print(f"Realm: {realm.realm}, Enabled: {realm.enabled}")

get_required_action async

get_required_action(realm_name, action_alias, namespace)

Get a specific required action by alias.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
action_alias	str	Alias of the required action	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
RequiredActionProviderRepresentation | None	Required action provider or None if not found

get_required_actions async

get_required_actions(realm_name, namespace)

Get all required actions for a realm.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
list[RequiredActionProviderRepresentation]	List of required action provider representations

get_resource_server_settings async

get_resource_server_settings(
    realm_name, client_uuid, namespace="default"
)

Get authorization resource server settings for a client.

API: GET /admin/realms/{realm}/clients/{id}/authz/resource-server

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client (not clientId)	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
dict[str, Any] | None	Resource server settings dict or None if not found

get_service_account_user async

get_service_account_user(
    client_uuid, realm_name, namespace
)

Get the service account user for a client.

Based on OpenAPI spec: GET /admin/realms/{realm}/clients/{id}/service-account-user

Parameters:

Name	Type	Description	Default
client_uuid	str	Client UUID in Keycloak	required
realm_name	str	Target realm name	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
UserRepresentation	Service account user representation as UserRepresentation

Raises:

Type	Description
KeycloakAdminError	If retrieval fails or service account is disabled

Example

user = await admin_client.get_service_account_user(client_uuid, "my-realm", "default") print(f"Service account user: {user.username}, ID: {user.id}")

get_user_federation_mappers async

get_user_federation_mappers(
    realm_name, parent_id, namespace="default"
)

Get all mappers for a user federation provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
parent_id	str	ID of the parent federation provider	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
list[ComponentRepresentation]	List of mapper configurations

get_user_federation_provider async

get_user_federation_provider(
    realm_name, provider_id, namespace="default"
)

Get a specific user federation provider by ID.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
provider_id	str	ID of the federation provider component	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
ComponentRepresentation | None	Federation provider configuration or None if not found

get_user_federation_provider_by_name async

get_user_federation_provider_by_name(
    realm_name, provider_name, namespace="default"
)

Get a user federation provider by its display name.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
provider_name	str	Display name of the federation provider	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
ComponentRepresentation | None	Federation provider configuration or None if not found

get_user_federation_providers async

get_user_federation_providers(
    realm_name, namespace="default"
)

Get all user federation providers for a realm.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
list[ComponentRepresentation]	List of user federation provider configurations

Example

providers = await admin_client.get_user_federation_providers("my-realm") for provider in providers: print(f"Provider: {provider.name}, Type: {provider.provider_id}")

link_organization_identity_provider async

link_organization_identity_provider(
    realm_name, org_id, idp_alias, namespace="default"
)

Link an identity provider to an organization.

API: POST /admin/realms/{realm}/organizations/{orgId}/identity-providers

Note: The redirect_uri is configured on the IdP itself, not in this link. Organizations link to existing realm IdPs by their alias.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
org_id	str	UUID of the organization	required
idp_alias	str	Alias of the identity provider in the realm	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if linked successfully or already linked

regenerate_client_secret async

regenerate_client_secret(client_id, realm_name, namespace)

Regenerate the client secret for a confidential client.

Parameters:

Name	Type	Description	Default
client_id	str	The client ID	required
realm_name	str	Name of the realm (defaults to "master")	required

Returns:

Type	Description
str | None	New client secret if successful, None otherwise

register_required_action async

register_required_action(
    realm_name, provider_id, name, namespace
)

Register a new required action provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
provider_id	str	Provider ID for the required action	required
name	str	Display name for the action	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

remove_client_default_scope async

remove_client_default_scope(
    realm_name, client_uuid, scope_id, namespace="default"
)

Remove a client scope from a client's default scopes.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	UUID of the client	required
scope_id	str	ID of the client scope to remove	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

remove_client_optional_scope async

remove_client_optional_scope(
    realm_name, client_uuid, scope_id, namespace="default"
)

Remove a client scope from a client's optional scopes.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	UUID of the client	required
scope_id	str	ID of the client scope to remove	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

remove_default_group async

remove_default_group(
    realm_name, group_id, namespace="default"
)

Remove a group from the default groups.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_id	str	ID of the group to remove from defaults	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

remove_realm_default_client_scope async

remove_realm_default_client_scope(
    realm_name, scope_id, namespace="default"
)

Remove a client scope from realm default client scopes.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope to remove	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

remove_realm_optional_client_scope async

remove_realm_optional_client_scope(
    realm_name, scope_id, namespace="default"
)

Remove a client scope from realm optional client scopes.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope to remove	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

remove_realm_role_composites async

remove_realm_role_composites(
    realm_name, role_name, child_roles, namespace="default"
)

Remove composite (child) roles from a realm role.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
role_name	str	Name of the parent role	required
child_roles	list[RoleRepresentation] | list[dict[str, Any]]	List of roles to remove from composites	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

remove_realm_roles_from_group async

remove_realm_roles_from_group(
    realm_name, group_id, roles, namespace="default"
)

Remove realm roles from a group.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_id	str	ID of the group	required
roles	list[RoleRepresentation] | list[dict[str, Any]]	List of roles to remove	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

test_ldap_authentication async

test_ldap_authentication(
    realm_name, connection_config, namespace="default"
)

Test LDAP authentication (bind) settings.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
connection_config	dict[str, Any]	LDAP connection configuration with bind credentials	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
dict[str, Any]	Dictionary with test results

test_ldap_connection async

test_ldap_connection(
    realm_name, connection_config, namespace="default"
)

Test LDAP connection settings before creating a provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
connection_config	dict[str, Any]	LDAP connection configuration to test	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
dict[str, Any]	Dictionary with test results

Example

result = await admin_client.test_ldap_connection( "my-realm", { "connectionUrl": "ldap://ldap.example.com:389", "bindDn": "cn=admin,dc=example,dc=org", "bindCredential": "secret", } ) if result.get("status") == "success": print("Connection successful!")

trigger_user_federation_sync async

trigger_user_federation_sync(
    realm_name,
    provider_id,
    full_sync=False,
    namespace="default",
)

Trigger synchronization for a user federation provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
provider_id	str	ID of the federation provider	required
full_sync	bool	If True, perform full sync; otherwise, sync only changed users	False
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
dict[str, Any]	Dictionary with sync results (added, updated, removed, failed counts)

Example

result = await admin_client.trigger_user_federation_sync( "my-realm", provider_id, full_sync=True ) print(f"Added: {result.get('added')}, Updated: {result.get('updated')}")

unlink_organization_identity_provider async

unlink_organization_identity_provider(
    realm_name, org_id, idp_alias, namespace="default"
)

Unlink an identity provider from an organization.

API: DELETE /admin/realms/{realm}/organizations/{orgId}/identity-providers/{alias}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
org_id	str	UUID of the organization	required
idp_alias	str	Alias of the identity provider to unlink	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if unlinked successfully or was not linked

update_authenticator_config async

update_authenticator_config(
    realm_name, config_id, config, namespace
)

Update an authenticator configuration.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
config_id	str	ID of the config to update	required
config	AuthenticatorConfigRepresentation | dict[str, Any]	Updated authenticator configuration	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

update_authorization_permission async

update_authorization_permission(
    realm_name,
    client_uuid,
    permission_type,
    permission_id,
    permission_data,
    namespace="default",
)

Update an authorization permission.

API: PUT /admin/realms/{realm}/clients/{id}/authz/resource-server/permission/{permission_type}/{permission_id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
permission_type	str	Type of permission ('resource' or 'scope')	required
permission_id	str	UUID of the permission to update	required
permission_data	dict	Updated permission configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful

update_authorization_policy async

update_authorization_policy(
    realm_name,
    client_uuid,
    policy_type,
    policy_id,
    policy_data,
    namespace="default",
)

Update an authorization policy.

API: PUT /admin/realms/{realm}/clients/{id}/authz/resource-server/policy/{type}/{policy-id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
policy_type	str	Policy type (role, user, group, client, time, js, aggregate, regex)	required
policy_id	str	UUID of the policy to update	required
policy_data	dict	Updated policy configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if updated successfully

update_authorization_resource async

update_authorization_resource(
    realm_name,
    client_uuid,
    resource_id,
    resource,
    namespace="default",
)

Update an authorization resource.

API: PUT /admin/realms/{realm}/clients/{id}/authz/resource-server/resource/{resource-id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
resource_id	str	UUID of the resource	required
resource	dict[str, Any]	Updated resource configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful

update_authorization_scope async

update_authorization_scope(
    realm_name,
    client_uuid,
    scope_id,
    scope,
    namespace="default",
)

Update an authorization scope.

API: PUT /admin/realms/{realm}/clients/{id}/authz/resource-server/scope/{scope-id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
scope_id	str	UUID of the scope	required
scope	dict[str, Any]	Updated scope configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful

update_client async

update_client(
    client_uuid, client_config, realm_name, namespace
)

Update an existing client configuration.

Parameters:

Name	Type	Description	Default
client_uuid	str	The UUID of the client to update	required
client_config	ClientRepresentation | dict[str, Any]	Updated client configuration as ClientRepresentation or dict	required
realm_name	str	Name of the realm	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful

Raises:

Type	Description
KeycloakAdminError	If client update fails

update_client_policies async

update_client_policies(
    realm_name, policies, namespace="default"
)

Update client policies for a realm.

This replaces ALL realm-level policies with the provided list. Global policies cannot be modified through this API.

API: PUT /admin/realms/{realm}/client-policies/policies

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
policies	list[dict[str, Any]]	List of policy configurations to set	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

Example

policies = [ { "name": "enforce-pkce-for-public", "description": "Enforce PKCE for public clients", "enabled": True, "conditions": [ {"condition": "client-access-type", "configuration": {"type": ["public"]}} ], "profiles": ["fapi-baseline"] } ] success = await admin_client.update_client_policies("my-realm", policies)

update_client_profiles async

update_client_profiles(
    realm_name, profiles, namespace="default"
)

Update client profiles for a realm.

This replaces ALL realm-level profiles with the provided list. Global profiles cannot be modified through this API.

API: PUT /admin/realms/{realm}/client-policies/profiles

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
profiles	list[dict[str, Any]]	List of profile configurations to set	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

Example

profiles = [ { "name": "fapi-advanced", "description": "FAPI 2.0 Advanced Security Profile", "executors": [ {"executor": "pkce-enforcer", "configuration": {"auto-configure": "true"}} ] } ] success = await admin_client.update_client_profiles("my-realm", profiles)

update_client_protocol_mapper async

update_client_protocol_mapper(
    client_uuid,
    mapper_id,
    mapper_config,
    realm_name="master",
)

Update a protocol mapper for a client.

Parameters:

Name	Type	Description	Default
client_uuid	str	UUID of the client in Keycloak	required
mapper_id	str	ID of the protocol mapper	required
mapper_config	ProtocolMapperRepresentation | dict[str, Any]	Updated protocol mapper configuration as ProtocolMapperRepresentation or dict	required
realm_name	str	Name of the realm	'master'

Returns:

Type	Description
bool	True if successful, False otherwise

Example

mappers = admin_client.get_client_protocol_mappers(client_uuid, "my-realm") mapper = mappers[0] mapper.protocol = "saml" success = admin_client.update_client_protocol_mapper( client_uuid, mapper.id, mapper, "my-realm" )

update_client_role async

update_client_role(
    client_uuid, role_name, role_config, realm_name="master"
)

Update a role for a client.

Parameters:

Name	Type	Description	Default
client_uuid	str	UUID of the client in Keycloak	required
role_name	str	Name of the role to update	required
role_config	RoleRepresentation | dict[str, Any]	Updated role configuration as RoleRepresentation or dict	required
realm_name	str	Name of the realm	'master'

Returns:

Type	Description
bool	True if successful, False otherwise

Example

role = admin_client.get_client_role(client_uuid, "admin", "my-realm") role.description = "Updated description" success = admin_client.update_client_role( client_uuid, "admin", role, "my-realm" )

update_client_scope async

update_client_scope(
    realm_name, scope_id, scope_config, namespace="default"
)

Update an existing client scope.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope to update	required
scope_config	ClientScopeRepresentation | dict[str, Any]	Updated client scope configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if update successful, False otherwise

update_client_scope_protocol_mapper async

update_client_scope_protocol_mapper(
    realm_name,
    scope_id,
    mapper_id,
    mapper_config,
    namespace="default",
)

Update a protocol mapper in a client scope.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
scope_id	str	ID of the client scope	required
mapper_id	str	ID of the mapper to update	required
mapper_config	ProtocolMapperRepresentation | dict[str, Any]	Updated protocol mapper configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if update successful, False otherwise

update_execution_requirement async

update_execution_requirement(
    realm_name,
    flow_alias,
    execution_id,
    requirement,
    namespace,
)

Update the requirement level of an execution.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
flow_alias	str	Alias of the parent flow	required
execution_id	str	ID of the execution to update	required
requirement	str	New requirement level (REQUIRED, ALTERNATIVE, DISABLED, CONDITIONAL)	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

update_group async

update_group(
    realm_name, group_id, group_config, namespace="default"
)

Update an existing group.

Note: Keycloak's PUT endpoint ignores subGroups - they must be managed separately via create_subgroup(). Read-only fields (path, subGroupCount, access) are excluded from the update payload to prevent 500 errors.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
group_id	str	ID of the group to update	required
group_config	GroupRepresentation | dict[str, Any]	Updated group configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

update_identity_provider async

update_identity_provider(
    realm_name, alias, provider_config, namespace
)

Update an existing identity provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
alias	str	Identity provider alias	required
provider_config	IdentityProviderRepresentation | dict[str, Any]	Updated identity provider configuration	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

idp = await admin_client.get_identity_provider("my-realm", "github", "default") idp.enabled = False success = await admin_client.update_identity_provider( "my-realm", "github", idp, "default" )

update_identity_provider_mapper async

update_identity_provider_mapper(
    realm_name, alias, mapper_id, mapper, namespace
)

Update an existing mapper.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
alias	str	Identity provider alias	required
mapper_id	str	Mapper ID	required
mapper	IdentityProviderMapperRepresentation | dict[str, Any]	Updated mapper configuration	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

update_organization async

update_organization(
    realm_name,
    org_id,
    organization_data,
    namespace="default",
)

Update an organization.

API: PUT /admin/realms/{realm}/organizations/{id}

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
org_id	str	UUID of the organization	required
organization_data	dict	Updated organization configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful

update_realm async

update_realm(realm_name, realm_config, namespace)

Update realm configuration.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm to update	required
realm_config	RealmRepresentation | dict[str, Any]	Updated realm configuration as RealmRepresentation or dict	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
RealmRepresentation	Updated realm configuration as RealmRepresentation

Raises:

Type	Description
KeycloakAdminError	If realm update fails

update_realm_role async

update_realm_role(
    realm_name, role_name, role_config, namespace="default"
)

Update an existing realm-level role.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
role_name	str	Name of the existing role	required
role_config	RoleRepresentation | dict[str, Any]	Updated role configuration	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

update_realm_themes async

update_realm_themes(realm_name, themes, namespace)

Update realm theme configuration.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
themes	dict[str, Any]	Theme configuration	required

Returns:

Type	Description
bool	True if successful, False otherwise

update_required_action async

update_required_action(
    realm_name, action_alias, action_config, namespace
)

Update a required action configuration.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
action_alias	str	Alias of the required action to update	required
action_config	RequiredActionProviderRepresentation | dict[str, Any]	Updated action configuration	required
namespace	str	Origin namespace for rate limiting	required

Returns:

Type	Description
bool	True if successful, False otherwise

update_resource_server_settings async

update_resource_server_settings(
    realm_name, client_uuid, settings, namespace="default"
)

Update authorization resource server settings for a client.

API: PUT /admin/realms/{realm}/clients/{id}/authz/resource-server

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
client_uuid	str	Internal UUID of the client	required
settings	dict[str, Any]	Resource server settings (policyEnforcementMode, decisionStrategy, etc.)	required
namespace	str	Namespace for rate limiting	'default'

Returns:

Type	Description
bool	True if successful

update_user_federation_provider async

update_user_federation_provider(
    realm_name,
    provider_id,
    federation_config,
    namespace="default",
)

Update an existing user federation provider.

Parameters:

Name	Type	Description	Default
realm_name	str	Name of the realm	required
provider_id	str	ID of the federation provider to update	required
federation_config	ComponentRepresentation | dict[str, Any]	Updated federation configuration	required
namespace	str	Kubernetes namespace (for logging)	'default'

Returns:

Type	Description
bool	True if successful, False otherwise

KeycloakAdminError

KeycloakAdminError(
    message, status_code=None, response_body=None
)

Bases: Exception

Base exception for Keycloak Admin API errors.

body_preview

body_preview(limit=2048)

Return a truncated preview of the response body for logging.

api_create

api_create(resource_name, conflict_is_success=True)

Decorator for CREATE operations.

Handles common error cases: - 409 Conflict: Optionally treated as success (resource exists) - Other errors: Returns None

Parameters:

Name	Type	Description	Default
resource_name	str	Human-readable resource name for logging	required
conflict_is_success	bool	If True, 409 returns None gracefully (idempotent create)	True

api_delete

api_delete(resource_name, not_found_is_success=True)

Decorator for DELETE operations.

Handles common error cases: - 404 Not Found: Optionally treated as success (idempotent delete) - Other errors: Returns False

Parameters:

Name	Type	Description	Default
resource_name	str	Human-readable resource name for logging	required
not_found_is_success	bool	If True, 404 returns True (idempotent delete)	True

api_get_list

api_get_list(resource_name)

Decorator for GET list operations.

Handles common error cases: - Returns empty list on any exception - Logs errors with consistent format

Parameters:

Name	Type	Description	Default
resource_name	str	Human-readable resource name for logging (e.g., "client scopes")	required

api_get_single

api_get_single(resource_name)

Decorator for GET single resource operations.

Handles common error cases: - Returns None on any exception (resource not found or error) - Logs errors with consistent format

Parameters:

Name	Type	Description	Default
resource_name	str	Human-readable resource name for logging (e.g., "client scope")	required

api_update

api_update(resource_name, conflict_is_success=False)

Decorator for UPDATE operations.

Handles common error cases: - 409 Conflict: Optionally treated as success (idempotent add) - Other errors: Returns False - Logs errors with consistent format

Parameters:

Name	Type	Description	Default
resource_name	str	Human-readable resource name for logging	required
conflict_is_success	bool	If True, treat 409 Conflict as success (already exists)	False

clear_admin_client_cache async

clear_admin_client_cache()

Clear all cached admin clients. Used in tests and operator shutdown.

get_keycloak_admin_client async

get_keycloak_admin_client(
    keycloak_name,
    namespace,
    rate_limiter=None,
    verify_ssl=False,
)

Get or create a cached KeycloakAdminClient for a specific instance.

The client is cached per (keycloak_name, namespace, verify_ssl) tuple and reused across reconciliation cycles. Token refresh/expiry is handled automatically by _ensure_authenticated() before every API call.

Parameters:

Name	Type	Description	Default
keycloak_name	str	Name of the Keycloak instance	required
namespace	str	Namespace where the Keycloak instance exists	required
rate_limiter	RateLimiter | None	Optional rate limiter for API throttling	None
verify_ssl	bool	Whether to verify SSL certificates (default: False for development)	False

Returns:

Type	Description
KeycloakAdminClient	Configured KeycloakAdminClient instance (may be cached)

Validation utilities for the Keycloak operator.

This module provides validation functions for Keycloak resources, configurations, and operator settings. It includes:

• Resource specification validation
• Configuration consistency checks
• Security and best practice validations
• Cross-resource dependency validation

ValidationError

ValidationError(message, field=None)

Bases: Exception

Exception raised for validation failures.

_extract_version_from_image

_extract_version_from_image(image)

Extract version tag from container image reference.

Parameters:

Name	Type	Description	Default
image	str	Container image reference like "quay.io/keycloak/keycloak:26.4.0"	required

Returns:

Type	Description
str | None	Version string or None if no version tag found

_parse_kubernetes_quantity

_parse_kubernetes_quantity(quantity)

Parse a Kubernetes quantity string into a numeric value.

Parameters:

Name	Type	Description	Default
quantity	str	Kubernetes quantity string (e.g., "100m", "1Gi", "2")	required

Returns:

Type	Description
float	Numeric value in base units

Raises:

Type	Description
ValueError	If quantity format is invalid

_parse_version

_parse_version(version_string)

Parse a semantic version string into major, minor, patch tuple.

Parameters:

Name	Type	Description	Default
version_string	str	Version string like "25.0.1" or "26.4.0"	required

Returns:

Type	Description
tuple[int, int, int]	Tuple of (major, minor, patch) as integers

Raises:

Type	Description
ValueError	If version format is invalid

get_health_port

get_health_port(image, version_override=None)

Get the port to use for health check endpoints based on Keycloak version.

Parameters:

Name	Type	Description	Default
image	str	Container image reference	required
version_override	str | None	Optional explicit version string (e.g., "24.0.5") for custom images without version tags. Takes precedence over image tag detection.	None

Returns:

Type	Description
int	Port number for health endpoints (8080 for 24.x, 9000 for 25.x+)

supports_management_port

supports_management_port(image, version_override=None)

Check if a Keycloak image supports the separate management port (9000).

The management interface with separate port 9000 was introduced in Keycloak 25.0.0. Earlier versions (24.x) serve health endpoints on the main HTTP port (8080).

Parameters:

Name	Type	Description	Default
image	str	Container image reference like "quay.io/keycloak/keycloak:26.4.0"	required
version_override	str | None	Optional explicit version string (e.g., "24.0.5") for custom images without version tags. Takes precedence over image tag detection.	None

Returns:

Type	Description
bool	True if the version supports management port (25.0.0+), False otherwise.
bool	Returns True if version cannot be determined (assume modern version).

supports_tracing

supports_tracing(image, version_override=None)

Check if a Keycloak image supports built-in OpenTelemetry tracing.

Built-in OTEL tracing support (via Quarkus) was introduced in Keycloak 26.0.0. Earlier versions (24.x, 25.x) do not have native tracing support.

Parameters:

Name	Type	Description	Default
image	str	Container image reference like "quay.io/keycloak/keycloak:26.4.0"	required
version_override	str | None	Optional explicit version string (e.g., "25.0.0") for custom images without version tags. Takes precedence over image tag detection.	None

Returns:

Type	Description
bool	True if the version supports tracing (26.0.0+), False otherwise.
bool	Returns True if version cannot be determined (assume modern version).

validate_client_id

validate_client_id(client_id)

Validate Keycloak client ID format.

Parameters:

Name	Type	Description	Default
client_id	str	Client ID to validate	required

Raises:

Type	Description
ValidationError	If client ID is invalid

validate_complete_resource

validate_complete_resource(
    resource, resource_type, namespace
)

Perform complete validation of a resource specification.

Parameters:

Name	Type	Description	Default
resource	dict[str, Any]	Complete resource definition	required
resource_type	str	Type of resource	required
namespace	str	Target namespace	required

Returns:

Type	Description
list[tuple[str, str, str]]	List of dependencies found during validation

Raises:

Type	Description
ValidationError	If resource is invalid

validate_cross_resource_references

validate_cross_resource_references(
    resource_spec, resource_type, namespace
)

Validate cross-resource references and return list of dependencies.

Parameters:

Name	Type	Description	Default
resource_spec	dict[str, Any]	Resource specification	required
resource_type	str	Type of the resource being validated	required
namespace	str	Current namespace	required

Returns:

Type	Description
list[tuple[str, str, str]]	List of (resource_type, name, namespace) tuples for dependencies

validate_environment_variables

validate_environment_variables(env_vars)

Validate environment variable configuration.

Parameters:

Name	Type	Description	Default
env_vars	dict[str, Any]	Environment variables dictionary	required

Raises:

Type	Description
ValidationError	If environment variable configuration is invalid

validate_image_reference

validate_image_reference(image)

Validate container image reference format.

Parameters:

Name	Type	Description	Default
image	str	Container image reference	required

Raises:

Type	Description
ValidationError	If image reference is invalid

validate_keycloak_version

validate_keycloak_version(image)

Validate Keycloak version is supported by this operator.

Supported versions: 24.x, 25.x, 26.x - 24.x: Health endpoints on main HTTP port (8080) - 25.x+: Health endpoints on management port (9000) with KC_HTTP_MANAGEMENT_PORT

Parameters:

Name	Type	Description	Default
image	str	Container image reference	required

Raises:

Type	Description
ValidationError	If Keycloak version is not supported

validate_namespace_name

validate_namespace_name(namespace)

Validate Kubernetes namespace name.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace name to validate	required

Raises:

Type	Description
ValidationError	If namespace name is invalid

validate_no_keycloak_placeholders

validate_no_keycloak_placeholders(
    value, field_name="field"
)

Validate that a string value does not contain Keycloak environment variable placeholders.

Keycloak supports placeholder syntax like ${keycloak:secret-name:key} or ${env.VAR} for runtime variable substitution. However, this operator manages Keycloak through the Admin REST API, not through Keycloak's config file mechanism. Placeholders cannot be resolved and will be passed literally to Keycloak, causing unexpected behavior.

Parameters:

Name	Type	Description	Default
value	str	String value to check	required
field_name	str	Name of the field for error messages	'field'

Raises:

Type	Description
ValidationError	If placeholder patterns are detected

validate_realm_name

validate_realm_name(realm_name)

Validate Keycloak realm name format.

Parameters:

Name	Type	Description	Default
realm_name	str	Realm name to validate	required

Raises:

Type	Description
ValidationError	If realm name is invalid

validate_redirect_uris

validate_redirect_uris(redirect_uris)

Validate OAuth2 redirect URIs.

Parameters:

Name	Type	Description	Default
redirect_uris	list[str]	List of redirect URIs to validate	required

Raises:

Type	Description
ValidationError	If any URI is invalid

validate_resource_limits

validate_resource_limits(resources)

Validate Kubernetes resource limits and requests.

Parameters:

Name	Type	Description	Default
resources	dict[str, Any]	Resource specification dictionary	required

Raises:

Type	Description
ValidationError	If resource specification is invalid

validate_resource_name

validate_resource_name(name, resource_type='resource')

Validate Kubernetes resource name according to DNS-1123 subdomain rules.

Parameters:

Name	Type	Description	Default
name	str	Resource name to validate	required
resource_type	str	Type of resource for error messages	'resource'

Raises:

Type	Description
ValidationError	If name is invalid

validate_security_settings

validate_security_settings(spec, resource_type)

Validate security-related settings and provide recommendations.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	Resource specification	required
resource_type	str	Type of resource being validated	required

validate_spec_no_placeholders

validate_spec_no_placeholders(spec, resource_type)

Recursively validate that a resource spec does not contain Keycloak placeholders.

Parameters:

Name	Type	Description	Default
spec	dict[str, Any]	Resource specification dictionary	required
resource_type	str	Type of resource for error messages	required

Raises:

Type	Description
ValidationError	If placeholder patterns are detected in any string field

validate_url

validate_url(url, url_type='URL')

Validate URL format and basic security checks.

Parameters:

Name	Type	Description	Default
url	str	URL to validate	required
url_type	str	Type of URL for error messages	'URL'

Raises:

Type	Description
ValidationError	If URL is invalid

Observability

Prometheus metrics for the Keycloak operator.

This module provides comprehensive metrics collection for monitoring operator performance, resource reconciliation, and system health.

Metric naming conventions (per ADR-048 and issue #171): - All metrics use the keycloak_operator_ prefix for Prometheus discoverability - Labels are kept to operationally essential dimensions to limit cardinality - High-cardinality labels (resource names, instance names) are excluded

MetricsCollector

MetricsCollector()

Collects and manages metrics for the Keycloak operator.

Initialize metrics collector.

admin_session_expires property

admin_session_expires

Gauge for admin session expiry timestamps.

admin_sessions_active property

admin_sessions_active

Gauge for active admin sessions.

record_database_connection_test

record_database_connection_test(
    namespace, database_type, success, duration
)

Record database connection test results.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace of the resource	required
database_type	str	Type of database	required
success	bool	Whether the connection test succeeded	required
duration	float	Time taken for the test	required

record_leader_election_change

record_leader_election_change()

Record a leader election change event.

record_lease_renewal

record_lease_renewal(instance_id, success, duration)

Record a leader election lease renewal attempt.

Parameters:

Name	Type	Description	Default
instance_id	str	Unique identifier for this operator instance	required
success	bool	Whether the lease renewal succeeded	required
duration	float	Time taken for the renewal operation	required

record_rbac_validation

record_rbac_validation(
    source_namespace, target_namespace, success
)

Record RBAC validation attempt.

Parameters:

Name	Type	Description	Default
source_namespace	str	Source namespace of the operation	required
target_namespace	str	Target namespace of the operation	required
success	bool	Whether the validation succeeded	required

record_reconciliation_skip

record_reconciliation_skip(resource_type, namespace, name)

Record a skipped reconciliation due to generation match.

This is called when a resource is already reconciled at the current generation and in Ready state, avoiding redundant API calls.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource (e.g., 'keycloak', 'realm', 'client')	required
namespace	str	Namespace of the resource	required
name	str	Name of the resource (for logging only, not in metric labels)	required

track_reconciliation async

track_reconciliation(
    resource_type, namespace, name, operation="reconcile"
)

Context manager to track reconciliation operations.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource being reconciled	required
namespace	str	Namespace of the resource	required
name	str	Name of the resource (used for logging only, not in metrics)	required
operation	str	Type of operation being performed	'reconcile'

update_cnpg_cluster_status

update_cnpg_cluster_status(namespace, healthy)

Update CloudNativePG cluster status.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace of the cluster	required
healthy	bool	Whether the cluster is healthy	required

update_keycloak_instance_status

update_keycloak_instance_status(namespace, running)

Update Keycloak instance status.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace of the instance	required
running	bool	Whether the instance is running	required

update_leader_election_status

update_leader_election_status(instance_id, is_leader)

Update leader election status.

Parameters:

Name	Type	Description	Default
instance_id	str	Unique identifier for this operator instance	required
is_leader	bool	Whether this instance is currently the leader	required

update_resource_status

update_resource_status(
    resource_type, namespace, phase, count=1
)

Update the count of resources in a specific phase.

Parameters:

Name	Type	Description	Default
resource_type	str	Type of resource	required
namespace	str	Namespace of the resource	required
phase	str	Current phase of the resource	required
count	int	Number of resources (default: 1)	1

MetricsServer

MetricsServer(port=8081, host='0.0.0.0')

HTTP server for exposing Prometheus metrics.

Initialize metrics server.

Parameters:

Name	Type	Description	Default
port	int	Port to serve metrics on	8081
host	str	Host interface to bind to	'0.0.0.0'

__aenter__ async

__aenter__()

Async context manager entry.

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)

Async context manager exit.

_health_handler async

_health_handler(request)

Handle /health endpoint for operator health checks.

_healthz_handler async

_healthz_handler(request)

Handle /healthz endpoint for Kubernetes compatibility.

_metrics_handler async

_metrics_handler(request)

Handle /metrics endpoint for Prometheus scraping.

_ready_handler async

_ready_handler(request)

Handle /ready endpoint for readiness probes.

_setup_routes

_setup_routes()

Set up HTTP routes for the metrics server.

start async

start()

Start the metrics server.

stop async

stop()

Stop the metrics server.

get_metrics_registry

get_metrics_registry()

Get or create the global metrics registry.

Health check utilities for the Keycloak operator.

This module provides comprehensive health checking capabilities for monitoring operator and system component health.

HealthCheckResult dataclass

HealthCheckResult(
    name,
    status,
    message,
    details=None,
    duration=0.0,
    timestamp=0.0,
)

Result of a health check operation.

HealthChecker

HealthChecker(k8s_client=None)

Performs comprehensive health checks for the operator.

Initialize health checker.

Parameters:

Name	Type	Description	Default
k8s_client	ApiClient | None	Kubernetes API client	None

_check_crds_installed async

_check_crds_installed()

Check if required CRDs are installed.

_check_kubernetes_api async

_check_kubernetes_api()

Check Kubernetes API connectivity.

_check_operator_resources async

_check_operator_resources()

Check operator's own resources (deployment, service account, etc.).

_check_rbac_permissions async

_check_rbac_permissions()

Check if the operator has required RBAC permissions.

check_all async

check_all()

Run all health checks.

Returns:

Type	Description
dict[str, HealthCheckResult]	Dictionary of health check results

get_overall_health

get_overall_health(results)

Determine overall health status from individual check results.

Parameters:

Name	Type	Description	Default
results	dict[str, HealthCheckResult]	Dictionary of health check results	required

Returns:

Type	Description
str	Overall health status

to_dict

to_dict(results)

Convert health check results to dictionary format.

Parameters:

Name	Type	Description	Default
results	dict[str, HealthCheckResult]	Health check results	required

Returns:

Type	Description
dict[str, Any]	Dictionary representation