- Release Status
- Need help?
- Getting Started
- Usage Guide
- Exceptions
- Pagination
- Logging
- Configuration Reference
- Rate Limiting
- Building the SDK
- Contributing
This repository contains the Okta management SDK for Python. This SDK can be used in your server-side code to interact with the Okta management API and
- Create and update users with the Users API
- Add security factors to users with the Factors API
- Manage groups with the Groups API
- Manage applications with the Apps API
- Much more!
Requires Python version 3.7.0 or higher.
You can also learn more on the Okta + Python page in our documentation.
This library uses semantic versioning and follows Okta's Library Version Policy.
| Version | Status |
|---|---|
| 0.x | |
| 1.x | |
| 2.x | ✔️ Release |
The latest release can always be found on the releases page.
If you run into problems using the SDK, you can:
- Ask questions on the Okta Developer Forums
- Post issues on GitHub (for code errors)
To install the Okta Python SDK in your project:
pip install oktaYou'll also need
- An Okta account, called an organization (sign up for a free developer organization if you need one)
- An API token
Construct a client instance by passing it your Okta domain name and API token:
import asyncio
from okta.client import Client as OktaClient
# Instantiating with a Python dictionary in the constructor
config = {
'orgUrl': 'https://{yourOktaDomain}',
'token': 'YOUR_API_TOKEN'
}
okta_client = OktaClient(config)
# example of usage, list all users and print their first name and last name
async def main():
users, resp, err = await okta_client.list_users()
for user in users:
print(user.profile.first_name, user.profile.last_name)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Another way to instantiate okta_client (config should be provided in other place as described below):
# Instantiating without in-text credentials
okta_client = OktaClient()Http session was introduced within v2.3.0 to allow custom SSL contest. Starting with SDK v2.4.0 you can reuse http session to gain better performance:
import asyncio
import aiohttp
from okta.client import Client as OktaClient
config = {
'orgUrl': 'https://{yourOktaDomain}',
'token': 'YOUR_API_TOKEN'
}
async def main():
async with OktaClient(config) as client:
# perform all queries within same session
users, okta_resp, err = await client.list_users()
user, okta_resp, err = await client.get_user(users[0].id)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Context Manager is a preferable way to use OktaClient to perform bunch of API requests.
Using a Python dictionary to hard-code the Okta domain and API token is encouraged for development; In production, you should use a more secure way of storing these values. This library supports a few different configuration sources, covered in the configuration reference section.
Okta allows you to interact with Okta APIs using scoped OAuth 2.0 access tokens. Each access token enables the bearer to perform specific actions on specific Okta endpoints, with that ability controlled by which scopes the access token contains.
This SDK supports this feature (OAuth 2.0) only for service-to-service applications. Check out our guides to learn more about how to register a new service application using a private and public key pair.
When using this approach you won't need an API Token because the SDK will request an access token for you. In order to use OAuth 2.0, construct a client instance by passing the following parameters:
from okta.client import Client as OktaClient
config = {
'orgUrl': 'https://{yourOktaDomain}',
'authorizationMode': 'PrivateKey',
'clientId': '{yourClientId}',
'scopes': ['okta.users.manage'],
'privateKey': 'YOUR_PRIVATE_JWK', # this parameter should be type of str
'kid': 'YOUR_PRIVATE_KEY_ID' # if a key ID needs to be provided, it can be provided here or part of the privateKey under "kid"
}
okta_client = OktaClient(config)
# example of usage, list all users and print their first name and last name
async def main():
users, resp, err = await okta_client.list_users()
for user in users:
print(user.profile.first_name, user.profile.last_name)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Note, that privateKey can be passed in JWK format or in PEM format, i.e. (examples generated with https://mkjwk.org):
{
"p": "4VmEO2ztlIHvalMHX797rWhETbKgB6bRbdGevYpRLZH167hKHB5vsuRjIAXJdujQw8W3rnas9Z-_Ddv1TbR5Qnz0UmhnxQAIbdDDUE9r5P_LEholrjY9Jz0P-W4jey-7cDATeITYHb3t67HcIwVbxQF5fkRdJAhfO029RqkH3OE",
"kty": "RSA",
"q": "x8ngsUMrDGReVVpeGdlZzGTSFxrNP89DF4WEQZ7zCpSe3_GpuUPbzgslYQEiX6XJY5ssavavVNOmmQEAt0xsMcxxVOPYCYy7LBE8cJQiFb_bMf2H1-zTlPn_KF4D10h45cLXhu-xh4c52Rh9WDMYZmKWLkAJQ6L_eueGoZkIDmU",
"d": "R38UamnZiEhOLxD7FYUN5AKj9mHQneRWizblxfNq2T1Nfk4matfZrrlMq_nz9tYZ3-TOCu3u-7k_igM0Tml365mbU_HzkfCrD-ou7cGSrqNgnipj_VQSgJfKRFKATEf4hMfdpKSd4rZzf8OJnq8s-kpRVC4kdHJtJjja59VvHEQRIrN_dkycNHSBWu5UjZbXOO5X3mjwuIh9gpLGZ-nHTqgTpT324q5BLVsH8_ywRGifIj-HQL1O5bJO2Q2_18iL1TbnMSbDwrKdb1edb4bgDuWB4o0xSTXsherTgeXu76gN9FY28tuAKSd34yqp7GZaYcjtkskbWPRtYhOID2cOgQ",
"e": "AQAB",
"use": "sig",
"kid": "test",
"qi": "FZGFuvW1W9VF31JyrMYJy_BH7vja3d9iZlhFzttNZ-wmiXG4irrI_fLJgmXK6dI3MfIhKPAYi9nnza2kcR1qEV9QObA4NV86RWnc8sAHbDGooe9VK5eJ5jjD7Tq_ZZiLiHGOZit3HylNilOb0k3VsgMcp0F3ZQaMbg35K9rSgZE",
"dp": "i4D6HjupvCTQDNdHmluU-d2xYxQwg2we_EgnaBkHdhmEzx8wKcYhyfIe90T92jH4gymUM1neatQw1yiS7D7MTn_CVH2zt730ed8h-kageYxsr1EmgHmtU-w2RmiLaIg9Fg99Dj_W9lqMvjtGFxwLGqN2DdfOfS79nV3bzbF4X6E",
"alg": "RS256",
"dq": "CT79iBacsmkeuIKDIl0du8jatDkIULCt4TPLqCHMC6xPIfwUJ7_NN17qru-XgKeyh0qSJq0d9iYJasFSICmIRFG62PvmbqK1stdlXaxtW2ZSpaCfHc4XCKj9NwgK03bGKZP314XWSHhoo_RvMJrEwVBEtQU_qIKtoil-4JGtfsU",
"n": "r95K3WIN8-4dB-tEKHjyTIIZZUMbHz8ad5oBX2BGiGxfPGfHbz2RH4QLT9ffzL-tgEo8IKs0Myh0VTwauiwz0cdHuS2gUTasK9OsosX1h1scSu_eZ-g-__lXBogU-SvBXBAgjv8hdcZjqWYQwmhJp2Ilv0CuXKxQwZyjso775PDjWDCH5HkVcSxHyUvpThLfWfkfz5PNDZvRpuPltv55ILRaVZhwPb7VXLAm2ebfeYUdybUKpGnEogKQdaL7TdNvP-HRnUSXTiYeXWHzU04FaXJ7yLmtXOQ52FT9dwkwLrCDOmDSBGafZ9asUtgOKhKN6wQW5mndhMK_1zThfjZyxQ"
}
or
-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCv3krdYg3z7h0H
60QoePJMghllQxsfPxp3mgFfYEaIbF88Z8dvPZEfhAtP19/Mv62ASjwgqzQzKHRV
PBq6LDPRx0e5LaBRNqwr06yixfWHWxxK795n6D7/+VcGiBT5K8FcECCO/yF1xmOp
ZhDCaEmnYiW/QK5crFDBnKOyjvvk8ONYMIfkeRVxLEfJS+lOEt9Z+R/Pk80Nm9Gm
4+W2/nkgtFpVmHA9vtVcsCbZ5t95hR3JtQqkacSiApB1ovtN028/4dGdRJdOJh5d
YfNTTgVpcnvIua1c5DnYVP13CTAusIM6YNIEZp9n1qxS2A4qEo3rBBbmad2Ewr/X
NOF+NnLFAgMBAAECggEAR38UamnZiEhOLxD7FYUN5AKj9mHQneRWizblxfNq2T1N
fk4matfZrrlMq/nz9tYZ3+TOCu3u+7k/igM0Tml365mbU/HzkfCrD+ou7cGSrqNg
nipj/VQSgJfKRFKATEf4hMfdpKSd4rZzf8OJnq8s+kpRVC4kdHJtJjja59VvHEQR
IrN/dkycNHSBWu5UjZbXOO5X3mjwuIh9gpLGZ+nHTqgTpT324q5BLVsH8/ywRGif
Ij+HQL1O5bJO2Q2/18iL1TbnMSbDwrKdb1edb4bgDuWB4o0xSTXsherTgeXu76gN
9FY28tuAKSd34yqp7GZaYcjtkskbWPRtYhOID2cOgQKBgQDhWYQ7bO2Uge9qUwdf
v3utaERNsqAHptFt0Z69ilEtkfXruEocHm+y5GMgBcl26NDDxbeudqz1n78N2/VN
tHlCfPRSaGfFAAht0MNQT2vk/8sSGiWuNj0nPQ/5biN7L7twMBN4hNgdve3rsdwj
BVvFAXl+RF0kCF87Tb1GqQfc4QKBgQDHyeCxQysMZF5VWl4Z2VnMZNIXGs0/z0MX
hYRBnvMKlJ7f8am5Q9vOCyVhASJfpcljmyxq9q9U06aZAQC3TGwxzHFU49gJjLss
ETxwlCIVv9sx/YfX7NOU+f8oXgPXSHjlwteG77GHhznZGH1YMxhmYpYuQAlDov96
54ahmQgOZQKBgQCLgPoeO6m8JNAM10eaW5T53bFjFDCDbB78SCdoGQd2GYTPHzAp
xiHJ8h73RP3aMfiDKZQzWd5q1DDXKJLsPsxOf8JUfbO3vfR53yH6RqB5jGyvUSaA
ea1T7DZGaItoiD0WD30OP9b2Woy+O0YXHAsao3YN1859Lv2dXdvNsXhfoQKBgAk+
/YgWnLJpHriCgyJdHbvI2rQ5CFCwreEzy6ghzAusTyH8FCe/zTde6q7vl4CnsodK
kiatHfYmCWrBUiApiERRutj75m6itbLXZV2sbVtmUqWgnx3OFwio/TcICtN2ximT
99eF1kh4aKP0bzCaxMFQRLUFP6iCraIpfuCRrX7FAoGAFZGFuvW1W9VF31JyrMYJ
y/BH7vja3d9iZlhFzttNZ+wmiXG4irrI/fLJgmXK6dI3MfIhKPAYi9nnza2kcR1q
EV9QObA4NV86RWnc8sAHbDGooe9VK5eJ5jjD7Tq/ZZiLiHGOZit3HylNilOb0k3V
sgMcp0F3ZQaMbg35K9rSgZE=
-----END PRIVATE KEY-----
Using a Python dictionary to hard-code the Okta domain and API token is encouraged for development; In production, you should use a more secure way of storing these values. This library supports a few different configuration sources, covered in the configuration reference section.
When creating a new client, we allow for you to pass custom instances of okta.request_executor, okta.http_client and okta.cache.cache.
from okta.client import Client as OktaClient
# Assuming implementations are in project.custom
from project.custom.request_executor_impl import RequestExecImpl
from project.custom.http_client_impl import HTTPClientImpl
from project.custom.cache_impl import CacheImpl
config = {
'orgUrl': 'https://{yourOktaDomain}',
'token': 'YOUR_API_TOKEN',
'requestExecutor': RequestExecImpl,
'httpClient': HTTPClientImpl,
'cacheManager': CacheImpl(), # pass instance of CacheImpl
'cache': {'enabled': True}
}
async def main():
client = OktaClient(config)
user_info, resp, err = await client.get_user({YOUR_USER_ID})
print(user_info)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Example: You can create a custom cache driver by implementing okta.cache.cache
# Fully working example for Custom Cache class
from okta.cache.cache import Cache
class CacheImpl(Cache):
def __init__(self):
super().__init__()
self.cache_dict = {}
def add(self, key, value):
self.cache_dict[key] = value
def get(self, key):
return self.cache_dict.get(key, None)
def contains(self, key):
return key in self.cache_dict
def delete(self, key):
if self.contains(key):
del self.cache_dict[key]A similar approach can be used to extend okta.request_executor:
from okta.request_executor import RequestExecutor
class RequestExecImpl(RequestExecutor):
def __init__(self, config, cache, http_client=None):
super().__init__(config, cache, http_client)
# custom code
# Note, this method shoud be defined as async
async def create_request(self, method: str, url: str, body: dict = None,
headers: dict = {}, oauth=False):
"""
Creates request for request executor's HTTP client.
Args:
method (str): HTTP Method to be used
url (str): URL to send request to
body (dict, optional): Request body. Defaults to None.
headers (dict, optional): Request headers. Defaults to {}.
Returns:
dict, Exception: Tuple of Dictionary repr of HTTP request and
exception raised during execution
"""
# custom code
# Note, this method shoud be defined as async
async def execute(self, request, response_type=None):
"""
This function is the high level request execution method. Performs the
API call and returns a formatted response object
Args:
request (dict): dictionary object containing request details
Returns:
(OktaAPIResponse, Exception): Response obj for the Okta API, Error
"""
# custom codeand okta.http_client:
from okta.http_client import HTTPClient
class HTTPClientImpl(HTTPClient):
def __init__(self, http_config={}):
super().__init__(http_config)
# custom code
# Note, this method shoud be defined as async
async def send_request(self, request):
"""
This method fires HTTP requests
Arguments:
request {dict} -- This dictionary contains all information needed
for the request.
- HTTP method (as str)
- Headers (as dict)
- Request body (as dict)
Returns:
Tuple(RequestInfo, ClientResponse, JSONBody, ErrorObject)
-- A tuple containing the request and response of the HTTP call
"""
# custom codeThese examples will help you understand how to use this library.
Once you initialize a client, you can call methods to make requests to the Okta API. The client uses asynchronous methods to operate. Most methods are grouped by the API endpoint they belong to. For example, methods that call the Users API are organized under the User resource client (okta.resource_clients.user_client.py).
Asynchronous I/O is fairly new to Python after making its debut in Python 3.5. It's powered by the
asynciolibrary which provides avenues to produce concurrent code. This allows developers to defineasyncfunctions andawaitasynchronous calls within them. For more information, you can check out the Python docs.
Calls using await must be made in an async def function. That function must be called by asyncio (see example below).
from okta.client import Client as OktaClient
import asyncio
async def main():
client = OktaClient()
users, resp, err = await client.list_users()
print(len(users))
loop = asyncio.get_event_loop()
loop.run_until_complete(main())This library should only be used with the Okta management API. To call the Authentication API, you should construct your own HTTP requests.
Assume the client is instantiated before each example below.
from okta.client import Client as OktaClient import okta.models as models client = OktaClient({'orgUrl': 'https://test.okta.com', 'token': 'YOUR_API_TOKEN'})
Custom attributes must first be defined in the Okta profile editor. Then, you can work with custom attributes on a user:
Feature is fully supported with SDK version >= 1.2.0
""" Setting attributes """
# Creating an instance through a Python Dictionary
from okta.models import UserProfile
user_profile = UserProfile({
'firstName': 'John',
'lastName': 'Foe',
'email': '[email protected]',
'login': '[email protected]',
'customAttr': 'custom value'
})
print(user_profile.customAttr)
# Creating an empty object and using variables
user_profile = models.UserProfile()
user_profile.first_name = 'John'
user_profile.last_name = 'Doe'
user_profile.email = '[email protected]'
user_profile.login = '[email protected]'
user_profile.customAttr = 'custom value'NOTE: Custom Attributes case must be the same as defined in the Profile tab of your organization UI.
To maintain consistent code, you can use camelCase for all default attributes:
Feature is fully supported with SDK version >= 1.6.1
user_profile = models.UserProfile()
user_profile.firstName = 'John'
user_profile.lastName = 'Doe'
user_profile.email = '[email protected]'
user_profile.login = '[email protected]'
user_profile.customAttr = 'custom value'Full example:
from okta.client import Client as OktaClient
import asyncio
async def main():
client = OktaClient()
# create user with custom attribute
body = {
"profile": {
"firstName": "John",
"lastName": "Smith",
"email": "[email protected]",
"login": "[email protected]",
"customAttr": "custom value"
},
"credentials": {
"password" : { "value": "Knock*knock*neo*111" }
}
}
result = await client.create_user(body)
# create user without custom attribute
body = {
"profile": {
"firstName": "Neo",
"lastName": "Anderson",
"email": "[email protected]",
"login": "[email protected]"
},
"credentials": {
"password" : { "value": "Knock*knock*neo*111" }
}
}
result = await client.create_user(body)
users, resp, err = await client.list_users()
for user in users:
print(user.profile.first_name, user.profile.last_name)
try:
print(user.profile.customAttr)
except:
print('User has no customAttr')
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Output should look like the following (removed pre-existing users from output):
John Smith
custom value
Neo Anderson
User has no customAttrFeature appears in v1.3.0
It is possible to set custom headers, which will be sent with each request:
import asyncio
from okta.client import Client as OktaClient
async def main():
client = OktaClient()
# set custom headers
client.set_custom_headers({'Custom-Header': 'custom value'})
# perform different requests with custom headers
users, resp, err = await client.list_users()
for user in users:
print(user.profile.first_name, user.profile.last_name)
# clear all custom headers
client.clear_custom_headers()
# output should be: {}
print(client.get_custom_headers())
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Note, that custom headers will be overwritten with default headers with the same name. This doesn't allow breaking the client. Get default headers:
client.get_default_headers()user, resp, err = await client.get_user(user.id)
# OR using their login
user, resp, err = await client.get_user(user.profile.login)users, resp, err = await client.list_users()# Query parameters are optional on methods that can use them!
# Check the method definition for details on which query parameters are accepted.
query_parameters = {'filter': 'status eq "ACTIVE"'}
users, resp, err = await client.list_users(query_parameters)# Create Password
password = models.PasswordCredential({
'value': 'Password123'
})
# Create User Credentials
user_creds = models.UserCredentials({
'password': password
})
# Create User Profile and CreateUser Request
user_profile = models.UserProfile()
user_profile.first_name = 'John'
user_profile.last_name = 'Doe'
user_profile.email = 'John.Doe'
user_profile.login = 'John.Doe'
create_user_req = models.CreateUserRequest({
'credentials': user_creds,
'profile': user_profile
})
# Create User
user, resp, err = await client.create_user(create_user_req)# Assume user object saved to variable `user`
# Craft new profile and get user object
new_profile = user.profile
new_profile.nick_name = 'Oktanaut'
updated_user_obj = models.User({'profile': new_profile})
# Update User with new details
updated_user, _, err = await client.update_user(user.id, updated_user_obj)You must first deactivate the user, and then you can delete the user.
# Assuming user starts off with a status of 'ACTIVE'
# Deactivate
resp, err = await client.deactivate_or_delete_user(user.id)
# Then delete
resp, err = await client.deactivate_or_delete_user(user.id)users_groups, resp, err = await client.list_user_groups(user.id)# Create Group Model
group_profile = models.GroupProfile({
'name': 'Group-Test'
})
group_model = models.Group({
'profile': group_profile
})
# Create Group
group, resp, err = await client.create_group(group_model)resp, err = await client.add_user_to_group(group.id, user.id)supported_factors, resp, err = await client.list_supported_factors(user.id)# Create and enroll factor
sms_factor = models.SmsUserFactor({
'profile': models.SmsUserFactorProfile({
'phoneNumber': '+12345678901'
})
})
enrolled_factor, _, err = await client.enroll_factor(created_user.id, sms_factor)activate_factor_request = models.ActivateFactorRequest({
'passCode': '123456'
})
activated_factor, resp, err = await client.activate_factor(user.id, factor.id, activate_factor_request)verify_factor_request = models.ActivateFactorRequest({
'passCode': '123456'
})
verified_factor, resp, err = await client.activate_factor(user.id, factor.id, verify_factor_request)apps, resp, err = await client.list_applications()app, resp, err = await client.get_application(app.id)# Create SWA Application model and SWA Application in Okta
swa_app_settings_app = models.SwaApplicationSettingsApplication({
'buttonField': 'btn-login',
'passwordField': 'txt-box-password',
'usernameField': 'txt-box-username',
'url': 'https://example.com/login.html',
'loginUrlRegex': '^https://example.com/login.html$'
})
swa_app_settings = models.SwaApplicationSettings({
'app': swa_app_settings_app
})
swa_app_model = models.SwaApplication({
'label': 'SWA Test App',
'settings': swa_app_settings,
})
app, resp, err = await client.create_application(swa_app_model)There are 2 ways of creating custom attribute for Group Schema Profile:
- via UI of your ORG (Directory -> Profile Editor -> Groups)
- with the following request (create custom attribute with name "testCustomAttr"):
definition = {'custom':
{'id': '#custom',
'properties':
{'testCustomAttr':
{'description': 'Custom attribute for testing purposes',
'maxLength': 20,
'minLength': 1,
'permissions': [{'action': 'READ_WRITE',
'principal': 'SELF'}],
'required': False,
'title': 'Test Custom Attribute',
'type': 'string'},
'required': []},
'type': 'object'
}
}
resp, _, err = await client.update_group_schema({'definitions': definition})Update existing attribute:
# Get existing GroupSchema
resp, _, err = await client.get_group_schema()
# Set new title for custom attribute 'testCustomAttr'
resp.definitions.custom.properties['testCustomAttr']['title'] = 'New Title'
# Launch api request to update GroupSchema
resp, _, err = await client.update_group_schema(resp)Not every API endpoint is represented by a method in this library. You can call any Okta management API endpoint using this generic syntax:
# Example that doesn't return Object
request, error = await client.get_request_executor().create_request(
method='POST',
url='/api/v1/users/USER_ID_HERE/lifecycle/activate',
body={},
headers={},
oauth=False
)
response, error = await client.get_request_executor().execute(request, None)
response_body = response.get_body()
# Example that does return Object
request, error = await client.get_request_executor().create_request(
method='GET',
url='/api/v1/users/USER_ID_HERE',
body={},
headers={},
oauth=False
)
response, error = await client.get_request_executor().execute(request, models.User)
response_body = client.form_response_body(response.get_body())
user = response.get_type()(response_body)Some group of api calls requires no empty parameters in request body, i.e. it is not allowed to pass within body the following samples {"some_var": ""} or {"some_var": []}. Meanwhile, other group of api calls allows setting empty parameters, which is used to clear value of related parameter on the server-side. By default all api calls are being performed without "empty" parameters, i.e. all empty parameters are being removed automatically before actual request. Starting from v2.1.0 users can control this behavior with parameter keep_empty_params:
This feature is supported with SDK version >= 2.1.0
# Default behavior, parameter "badgeNumber" won't be present in actual request
from okta.client import Client as OktaClient
import asyncio
async def main():
client = OktaClient()
user_id = 'REDACTED'
user_params = {"profile": {"phoneNumber": "1234567890", "badgeNumber": ""}}
updated_user, resp, err = await client.partial_update_user(user_id, user_params)
print(updated_user)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())The same behavior when "keep_empty_params" is set to "False" (which is default for most methods, except UserSchema methods):
# Parameter "badgeNumber" won't be present in actual request
from okta.client import Client as OktaClient
import asyncio
async def main():
client = OktaClient()
user_id = 'REDACTED'
user_params = {"profile": {"phoneNumber": "1234567890", "badgeNumber": ""}}
updated_user, resp, err = await client.partial_update_user(user_id, user_params, keep_empty_params=False)
print(updated_user)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())All empty parameters will be present in request when "keep_empty_params" is set to "True":
# Parameter "badgeNumber" will be present in actual request
from okta.client import Client as OktaClient
import asyncio
async def main():
client = OktaClient()
user_id = 'REDACTED'
user_params = {"profile": {"phoneNumber": "1234567890", "badgeNumber": ""}}
updated_user, resp, err = await client.partial_update_user(user_id, user_params, keep_empty_params=True)
print(updated_user)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Starting from v1.1.0 SDK introduces exceptions, which are disabled by default, thus feature is backward compatible. To force client raise an exception instead of returning custom error, option 'raiseException' should be provided:
import asyncio
from okta.client import Client as OktaClient
from okta.exceptions import OktaAPIException
async def main():
config = {'orgUrl': 'https://{yourOktaDomain}',
'token': 'bad_token',
'raiseException': True}
client = OktaClient(config)
try:
users, resp, err = await client.list_users()
for user in users:
print(user.profile.first_name, user.profile.last_name)
except OktaAPIException as err:
print(err)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Result should look like:
{'errorCode': 'E0000011', 'errorSummary': 'Invalid token provided', 'errorLink': 'E0000011', 'errorId': 'oaeqWcqizEUQ_-iHc2hCbH9LA', 'errorCauses': []}List of available exceptions: OktaAPIException, HTTPException (to raise instead of returning errors OktaAPIError and HTTPError respectively). It is possible to inherit and/or extend given exceptions:
from okta.exceptions import HTTPException
class MyHTTPException(HTTPException):
pass
raise MyHTTPException('My HTTP Exception')If your request comes back with more than the default or set limit (resp.has_next() == True), you can request the next page.
Example of listing users 1 at a time:
query_parameters = {'limit': '1'}
users, resp, err = await client.list_users(query_parameters)
# Check if there more pages follow
if resp.has_next():
users, err = await resp.next() # Returns list of 1 user after the last retrieved user
# Iterate through all of the rest of the pages
while resp.has_next():
users, err = await resp.next()
# Do stuff with users in users
print(resp.has_next()) # False
try:
await resp.next()
except StopAsyncIteration:
# Handle Exception raisedHere's a complete example:
from okta.client import Client as OktaClient
import asyncio
async def main():
client = OktaClient()
users, resp, err = await client.list_users()
while True:
for user in users:
print(user.profile.login) # Add more properties here.
if resp.has_next():
users, err = await resp.next()
else:
break
loop = asyncio.get_event_loop()
loop.run_until_complete(main())If you require access to the response headers during a pagination operation, set the includeResponse parameter to True in the call to the next method; this returns the response as a third tuple value. See the following example:
from okta.client import Client as OktaClient
import asyncio
async def main():
client = OktaClient()
users, resp, err = await client.list_users()
while True:
for user in users:
print(user.profile.login)
if resp.has_next():
users, err, response = await resp.next(True) # Specify True to receive the response object as the third part of the tuple for further analysis
else:
break
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Feature appears in version 1.5.0
SDK v1.5.0 introduces logging for debug purposes. Logs are disabled by default, thus SDK behavior remains the same. Logging should be enabled explicitly via client configuration or via a configuration file:
from okta.client import Client as OktaClient
config = {"logging": {"enabled": True}}
client = OktaClient(config)SDK utilizes the standard Python library logging. By default, log level INFO is set. You can set another log level via config:
from okta.client import Client as OktaClient
import logging
config = {"logging": {"enabled": True, "logLevel": logging.DEBUG}}
client = OktaClient(config)NOTE: DO NOT SET DEBUG LEVEL IN PRODUCTION!
Here's a complete example:
from okta.client import Client as OktaClient
import asyncio
import logging
async def main():
config = {"logging": {"enabled": True, "logLevel": logging.DEBUG}}
client = OktaClient(config)
users, resp, err = await client.list_users()
assert users is not None
loop = asyncio.get_event_loop()
loop.run_until_complete(main())You should now see logs in your console. Actual API Tokens will be logged to the console, so use caution and never use DEBUG level logging in production.
What it being logged: requests, http errors, caching responses.
This library looks for configuration in the following sources:
- An
okta.yamlfile in a.oktafolder in the current user's home directory (~/.okta/okta.yamlor%userprofile%\.okta\okta.yaml). See a sample YAML Configuration - A
okta.yamlfile in the application or project's root directory. See a sample YAML Configuration - Environment variables
- Configuration explicitly passed to the constructor (see the example in Getting started)
Only ONE source needs to be provided!
Higher numbers win. In other words, configuration passed via the constructor will OVERRIDE configuration found in environment variables, which will override configuration in the designated okta.yaml files.
When you use an API Token instead of OAuth 2.0 the full YAML configuration looks like:
okta:
client:
connectionTimeout: 30 # seconds
orgUrl: "https://{yourOktaDomain}"
proxy:
port: { proxy_port }
host: { proxy_host }
username: { proxy_username }
password: { proxy_password }
token: "YOUR_API_TOKEN"
requestTimeout: 0 # seconds
rateLimit:
maxRetries: 4
logging:
enabled: true
logLevel: INFOWhen you use OAuth 2.0 the full YAML configuration looks like:
okta:
client:
connectionTimeout: 30 # seconds
orgUrl: "https://{yourOktaDomain}"
proxy:
port: { proxy_port }
host: { proxy_host }
username: { proxy_username }
password: { proxy_password }
authorizationMode: "PrivateKey"
clientId: "YOUR_CLIENT_ID"
scopes:
- scope.1
- scope.2
privateKey: |
-----BEGIN RSA PRIVATE KEY-----
MIIEogIBAAKCAQEAl4F5CrP6Wu2kKwH1Z+CNBdo0iteHhVRIXeHdeoqIB1iXvuv4
THQdM5PIlot6XmeV1KUKuzw2ewDeb5zcasA4QHPcSVh2+KzbttPQ+RUXCUAr5t+r
0r6gBc5Dy1IPjCFsqsPJXFwqe3RzUb...
-----END RSA PRIVATE KEY-----
requestTimeout: 0 # seconds
rateLimit:
maxRetries: 4
logging:
enabled: true
logLevel: INFOIf a proxy is not going to be used for the SDK, you may omit the
okta.client.proxysection from yourokta.yamlfile
Each one of the configuration values above can be turned into an environment variable name with the _ (underscore) character and UPPERCASE characters. The following are accepted:
OKTA_CLIENT_AUTHORIZATIONMODEOKTA_CLIENT_ORGURLOKTA_CLIENT_TOKENOKTA_CLIENT_CLIENTIDOKTA_CLIENT_SCOPESOKTA_CLIENT_PRIVATEKEYOKTA_CLIENT_USERAGENTOKTA_CLIENT_CONNECTIONTIMEOUTOKTA_CLIENT_REQUESTTIMEOUTOKTA_CLIENT_CACHE_ENABLEDOKTA_CLIENT_CACHE_DEFAULTTTIOKTA_CLIENT_CACHE_DEFAULTTTLOKTA_CLIENT_PROXY_PORTOKTA_CLIENT_PROXY_HOSTOKTA_CLIENT_PROXY_USERNAMEOKTA_CLIENT_PROXY_PASSWORDOKTA_CLIENT_RATELIMIT_MAXRETRIESOKTA_TESTING_TESTINGDISABLEHTTPSCHECK
Starting with SDK v2.3.0 you can provide custom SSL context:
import asyncio
import ssl
from okta.client import Client as OktaClient
async def main():
# create default context for demo purpose
ssl_context = ssl.create_default_context()
client = OktaClient({"sslContext": ssl_context})
users, resp, err = await client.list_users()
print(users)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())The Okta API will return 429 responses if too many requests are made within a given time. Please see Rate Limiting at Okta for a complete list of which endpoints are rate limited. When a 429 error is received, the X-Rate-Limit-Reset header will tell you the time at which you can retry. This section discusses the method for handling rate limiting with this SDK.
This SDK uses the built-in retry strategy to automatically retry on 429 errors. You can use the default configuration options for the built-in retry strategy, or provide your desired values via the client configuration.
You can configure the following options when using the built-in retry strategy:
| Configuration Option | Description |
|---|---|
| client.requestTimeout | The waiting time in seconds for a request to be resolved by the client. Less than or equal to 0 means "no timeout". The default value is 0 (None). |
| client.rateLimit.maxRetries | The number of times to retry. |
Check out the Configuration Reference section for more details about how to set these values via configuration.
In most cases, you won't need to build the SDK from source. If you want to build it yourself, you'll need these prerequisites:
- Clone the repo
- Run
python setup.py buildfrom the root of the project (assuming Python is installed) - Ensure tests run succesfully. Install
toxif not installed already using:pip install tox. Run tests usingtoxin the root directory of the project.
We're happy to accept contributions and PRs! Please see the Contribution Guide to understand how to structure a contribution.
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent