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.

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.

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.

KeycloakAuthenticationFlow

Bases: BaseModel

Authentication flow configuration.

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.

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.

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()

Convert the realm specification to Keycloak API format.

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

User federation configuration.

OperatorRef

Bases: BaseModel

Reference to the operator managing this realm.

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.

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.

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.

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.

ServiceAccountRoles

Bases: BaseModel

Role mappings for service account users.

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.

delete_keycloak_instance async

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

Handle Keycloak instance deletion with proper finalizer management.

This handler performs comprehensive cleanup of all associated resources and removes the finalizer only after cleanup is complete, preventing data loss and orphaned resources.

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

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).

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(
    spec, name, namespace, status, patch, **kwargs
)

Periodic health check for Keycloak instances.

This timer handler runs every 60 seconds to check the health
of Keycloak instances and update their status accordingly.

Args:
    spec: Keycloak resource specification
    name: Name of the Keycloak resource
    namespace: Namespace where the resource exists
    status: Current status of the resource

Returns:
    Dictionary with updated status information, or None if no changes

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.

_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, **kwargs
)

Handle KeycloakRealm deletion with proper finalizer management.

This handler performs comprehensive cleanup of the realm from Keycloak and any associated Kubernetes resources, removing the finalizer only after cleanup is complete.

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

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.

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(
    spec, name, namespace, status, patch, **_kwargs
)

Periodic health check for KeycloakRealms.

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

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

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.

delete_keycloak_client async

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

Handle KeycloakClient deletion with proper finalizer management.

This handler performs comprehensive cleanup of the client from Keycloak and any associated Kubernetes resources, removing the finalizer only after cleanup is complete.

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

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.

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(
    spec, name, namespace, status, patch, **kwargs
)

Periodic health check for KeycloakClients.

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

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

Returns:

Type	Description
None	Dictionary with updated status, or None if no changes needed

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

_create_backup async

_create_backup(name, namespace, keycloak_spec)

Create a backup of Keycloak data before deletion.

_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.

_store_keycloak_backup_in_secret async

_store_keycloak_backup_in_secret(
    backup_data, backup_name, namespace
)

Store full Keycloak instance backup data in a Kubernetes secret.

Parameters:

Name	Type	Description	Default
backup_data	dict[str, Any]	Complete Keycloak instance backup data	required
backup_name	str	Name for the backup	required
namespace	str	Namespace to store the secret	required

_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_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
bool	True if deployment became ready, False if timed out

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

_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

_create_realm_backup async

_create_realm_backup(
    name, namespace, realm_spec, admin_client
)

Create a backup of realm data before deletion.

_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

_store_backup_in_secret async

_store_backup_in_secret(
    backup_data, backup_name, namespace, backup_type
)

Store backup data in a Kubernetes secret.

Parameters:

Name	Type	Description	Default
backup_data	dict[str, Any]	Backup data to store	required
backup_name	str	Name for the backup	required
namespace	str	Namespace to store the secret	required
backup_type	str	Type of backup	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

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.

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.

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 (LDAP, Active Directory, etc.).

Parameters:

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

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.)	{}

manage_realm_backup async

manage_realm_backup(spec, name, namespace)

Manage realm backup operations based on spec configuration.

Parameters:

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

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

_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

_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_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
)

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

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

backup_keycloak_data

backup_keycloak_data(name, namespace, spec, k8s_client)

Create a Kubernetes Job to backup Keycloak data.

Parameters:

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

Returns:

Type	Description
V1Job	Created Job object

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,
)

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

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_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	Any	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_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_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,
)

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

__aenter__ async

__aenter__()

Async context manager entry.

__aexit__ async

__aexit__(exc_type, exc_val, exc_tb)

Async context manager exit - ensures session cleanup.

_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).

_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

_refresh_token async

_refresh_token()

Refresh the access token using the refresh token.

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_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.

backup_realm async

backup_realm(realm_name, namespace)

Create a backup of realm configuration.

Parameters:

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

Returns:

Type	Description
dict[str, Any] | None	Realm backup data if successful, None otherwise

close async

close()

Close method for compatibility with async context manager.

Note: With the caching strategy, we don't actually close the httpx client here as it's shared across multiple KeycloakAdminClient instances. The cached client will be reused until the operator shuts down.

configure_authentication_flow async

configure_authentication_flow(realm_name, flow_config)

Configure 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 as AuthenticationFlowRepresentation or dict	required

Returns:

Type	Description
bool	True if successful, False otherwise

Example

from keycloak_operator.models.keycloak_api import AuthenticationFlowRepresentation

flow = AuthenticationFlowRepresentation( alias="browser-custom", description="Custom browser flow" ) success = admin_client.configure_authentication_flow("my-realm", flow)

configure_identity_provider async

configure_identity_provider(
    realm_name, provider_config, namespace
)

Configure identity provider for a realm.

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_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)

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_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

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_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

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_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_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] | None	List of protocol mapper configurations as ProtocolMapperRepresentation or None on error

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] | None	List of client role configurations as RoleRepresentation or None on error

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_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_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_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_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_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}")

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

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_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_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_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

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.

get_keycloak_admin_client async

get_keycloak_admin_client(
    keycloak_name,
    namespace,
    rate_limiter=None,
    verify_ssl=False,
)

Factory function to create KeycloakAdminClient for a specific instance.

This function handles: - Looking up Keycloak instance details from Kubernetes - Retrieving admin credentials from secrets - Creating configured admin client with rate limiting

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

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

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 supports required features (management port).

The management interface with separate port 9000 was introduced in Keycloak 25.0.0. Earlier versions do not support KC_HTTP_MANAGEMENT_PORT and will fail health checks.

Parameters:

Name	Type	Description	Default
image	str	Container image reference	required

Raises:

Type	Description
ValidationError	If Keycloak version is too old and doesn't support management port

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_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_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.

MetricsCollector

MetricsCollector()

Collects and manages metrics for the Keycloak operator.

Initialize metrics collector.

authorization_failures property

authorization_failures

Counter for authorization failures.

operational_tokens_active property

operational_tokens_active

Gauge for active operational tokens.

token_expires property

token_expires

Gauge for token expiry timestamps.

record_database_connection_test

record_database_connection_test(
    resource_name,
    namespace,
    database_type,
    success,
    duration,
)

Record database connection test results.

Parameters:

Name	Type	Description	Default
resource_name	str	Name of the Keycloak resource	required
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(
    previous_leader, new_leader, namespace
)

Record a leader election change event.

Parameters:

Name	Type	Description	Default
previous_leader	str	ID of the previous leader	required
new_leader	str	ID of the new leader	required
namespace	str	Namespace where the election occurred	required

record_lease_renewal

record_lease_renewal(
    instance_id, namespace, success, duration
)

Record a leader election lease renewal attempt.

Parameters:

Name	Type	Description	Default
instance_id	str	Unique identifier for this operator instance	required
namespace	str	Namespace where the operator is running	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

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	required
operation	str	Type of operation being performed	'reconcile'

update_cnpg_cluster_status

update_cnpg_cluster_status(
    cluster_name, namespace, healthy
)

Update CloudNativePG cluster status.

Parameters:

Name	Type	Description	Default
cluster_name	str	Name of the CNPG cluster	required
namespace	str	Namespace of the cluster	required
healthy	bool	Whether the cluster is healthy	required

update_keycloak_instance_status

update_keycloak_instance_status(
    instance_name, namespace, running
)

Update Keycloak instance status.

Parameters:

Name	Type	Description	Default
instance_name	str	Name of the Keycloak instance	required
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, namespace, is_leader
)

Update leader election status.

Parameters:

Name	Type	Description	Default
instance_id	str	Unique identifier for this operator instance	required
namespace	str	Namespace where the operator is running	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