Use Python Keycloak Asynchronously¶
Asynchronous admin client¶
Configure admin client¶
admin = KeycloakAdmin(
server_url="http://localhost:8080/",
username='example-admin',
password='secret',
realm_name="master",
user_realm_name="only_if_other_realm_than_master")
Configure admin client with connection¶
from keycloak import KeycloakAdmin
from keycloak import KeycloakOpenIDConnection
keycloak_connection = KeycloakOpenIDConnection(
server_url="http://localhost:8080/",
username='example-admin',
password='secret',
realm_name="master",
user_realm_name="only_if_other_realm_than_master",
client_id="my_client",
client_secret_key="client-secret",
verify=True)
keycloak_admin = KeycloakAdmin(connection=keycloak_connection)
Create user asynchronously¶
new_user = await keycloak_admin.a_create_user({"email": "example@example.com",
"username": "example@example.com",
"enabled": True,
"firstName": "Example",
"lastName": "Example"})
Add user asynchronously and raise exception if username already exists¶
The exist_ok currently defaults to True for backwards compatibility reasons.
new_user = await keycloak_admin.a_create_user({"email": "example@example.com",
"username": "example@example.com",
"enabled": True,
"firstName": "Example",
"lastName": "Example"},
exist_ok=False)
Add user asynchronously and set password¶
new_user = await keycloak_admin.a_create_user({"email": "example@example.com",
"username": "example@example.com",
"enabled": True,
"firstName": "Example",
"lastName": "Example",
"credentials": [{"value": "secret","type": "password",}]})
Add user asynchronous and specify a locale¶
new_user = await keycloak_admin.a_create_user({"email": "example@example.fr",
"username": "example@example.fr",
"enabled": True,
"firstName": "Example",
"lastName": "Example",
"attributes": {
"locale": ["fr"]
}})
Asynchronous User counter¶
count_users = await keycloak_admin.a_users_count()
Get users Returns a list of users asynchronously, filtered according to query parameters¶
users = await keycloak_admin.a_get_users({})
Get user ID asynchronously from username¶
user_id_keycloak = await keycloak_admin.a_get_user_id("username-keycloak")
Get user asynchronously¶
user = await keycloak_admin.a_get_user("user-id-keycloak")
Update user asynchronously¶
response = await keycloak_admin.a_update_user(user_id="user-id-keycloak",
payload={'firstName': 'Example Update'})
Update user password asynchronously¶
response = await keycloak_admin.a_set_user_password(user_id="user-id-keycloak", password="secret", temporary=True)
Get user credentials asynchronously¶
credentials = await keycloak_admin.a_get_credentials(user_id='user_id')
Get user credential asynchronously by ID¶
credential = await keycloak_admin.a_get_credential(user_id='user_id', credential_id='credential_id')
Delete user credential asynchronously¶
response = await keycloak_admin.a_delete_credential(user_id='user_id', credential_id='credential_id')
Delete User asynchronously¶
response = await keycloak_admin.a_delete_user(user_id="user-id-keycloak")
Get consents granted asynchronously by the user¶
consents = await keycloak_admin.a_consents_user(user_id="user-id-keycloak")
Send user action asynchronously¶
response = await keycloak_admin.a_send_update_account(user_id="user-id-keycloak",
payload=['UPDATE_PASSWORD'])
Send verify email asynchronously¶
response = await keycloak_admin.a_send_verify_email(user_id="user-id-keycloak")
Get sessions associated asynchronously with the user¶
sessions = await keycloak_admin.a_get_sessions(user_id="user-id-keycloak")
Asynchronous OpenID Client¶
Asynchronous Configure client OpenID¶
from keycloak import KeycloakOpenID
# Configure client
# For versions older than 18 /auth/ must be added at the end of the server_url.
keycloak_openid = KeycloakOpenID(server_url="http://localhost:8080/",
client_id="example_client",
realm_name="example_realm",
client_secret_key="secret")
Get .well_know asynchronously¶
config_well_known = await keycloak_openid.a_well_known()
Get access token asynchronously with code¶
access_token = await keycloak_openid.a_token(
grant_type='authorization_code',
code='the_code_you_get_from_auth_url_callback',
redirect_uri="your_call_back_url")
Get access asynchronously token with user and password¶
token = await keycloak_openid.a_token("user", "password")
token = await keycloak_openid.a_token("user", "password", totp="012345")
Get token asynchronously using Token Exchange¶
token = await keycloak_openid.a_exchange_token(token['access_token'],
"my_client", "other_client", "some_user")
Refresh token asynchronously¶
token = await keycloak_openid.a_refresh_token(token['refresh_token'])
Get UserInfo asynchronously¶
userinfo = await keycloak_openid.a_userinfo(token['access_token'])
Logout asynchronously¶
await keycloak_openid.a_logout(token['refresh_token'])
Get certs asynchronously¶
certs = await keycloak_openid.a_certs()
Introspect RPT asynchronously¶
token_rpt_info = await keycloak_openid.a_introspect(await keycloak_openid.a_introspect(token['access_token'],
rpt=rpt['rpt'],
token_type_hint="requesting_party_token"))
Introspect token asynchronously¶
token_info = await keycloak_openid.a_introspect(token['access_token'])
Decode token asynchronously¶
token_info = await keycloak_openid.a_decode_token(token['access_token'])
# Without validation
token_info = await keycloak_openid.a_decode_token(token['access_token'], validate=False)
Get UMA-permissions asynchronously by token¶
token = await keycloak_openid.a_token("user", "password")
permissions = await keycloak_openid.a_uma_permissions(token['access_token'])
Get UMA-permissions asynchronously by token with specific resource and scope requested¶
token = await keycloak_openid.a_token("user", "password")
permissions = await keycloak_openid.a_uma_permissions(token['access_token'], permissions="Resource#Scope")
Get auth status asynchronously for a specific resource and scope by token¶
token = await keycloak_openid.a_token("user", "password")
auth_status = await keycloak_openid.a_has_uma_access(token['access_token'], "Resource#Scope")
Asynchronous UMA¶
Asynchronous Configure client UMA¶
from keycloak import KeycloakOpenIDConnection
from keycloak import KeycloakUMA
keycloak_connection = KeycloakOpenIDConnection(
server_url="http://localhost:8080/",
realm_name="master",
client_id="my_client",
client_secret_key="client-secret")
keycloak_uma = KeycloakUMA(connection=keycloak_connection)
Create a resource set asynchronously¶
resource_set = await keycloak_uma.a_resource_set_create({
"name": "example_resource",
"scopes": ["example:read", "example:write"],
"type": "urn:example"})
List resource sets asynchronously¶
resource_sets = await uma.a_resource_set_list()
Get resource set asynchronously¶
latest_resource = await uma.a_resource_set_read(resource_set["_id"])
Update resource set asynchronously¶
latest_resource["name"] = "New Resource Name"
await uma.a_resource_set_update(resource_set["_id"], latest_resource)
Delete resource set asynchronously¶
await uma.a_resource_set_delete(resource_id=resource_set["_id"])