GitHub - okta/okta-sdk-python: Python SDK for Okta Management APIs

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

This Python package is automatically generated by the OpenAPI Generator project:

Package version: 3.1.0
OpenAPI JAR Generator version: 7.7.0
Build package: org.openapitools.codegen.languages.PythonClientCodegen For more information, please visit https://developer.okta.com/

Requirements.

We have stopped supporting Python versions below 3.9.

Requires Python version 3.9.0 or higher.

You can also learn more on the [Okta + Python][lang-landing-page] page in our documentation.

Release status

This library uses semantic versioning and follows Okta's [Library Version Policy][okta-library-versioning].

Version	Status
2.x	⚠️ Deprecated
3.x	✔️ Release

The latest release can always be found on the [releases page][github-releases].

Need help?

If you run into problems using the SDK, you can:

Ask questions on the [Okta Developer Forums][devforum]
Post [issues on GitHub][github-issues] (for code errors)

Installation & Usage

pip install

If the repository is cloned, you can install directly using the below command at root directory (where setup.py is located):

To install the Okta Python SDK in your project:

You'll also need

An Okta account, called an organization (sign up for a free [developer organization][dev-okta-signup] if you need one)
An [API token][api-token-docs]

Construct a client instance by passing it your Okta domain name and API token:

Setuptools

Install via Setuptools.

python setup.py install --user

(or sudo python setup.py install to install the package for all users)

Then import the package:

Generate models and APIs

Models and APIs are auto-generated in accordance with specific openapi spec defined in openapi/api.yaml file. In order to make changes in models and/or APIs you need to determine what is the root of changes:

Prerequisites

This project uses the JAR version of openapi-generator-cli to resolve specific configuration issues found in the NPM package.

Install Java: Ensure you have a Java Runtime Environment (JRE) installed.
Download the Generator: Download the latest JAR file from the official documentation.
Setup: Place the downloaded JAR file into the openapi directory.
Generate: Run the generation script:

if change is needed because of new version of openapi spec that has been released, then:

Make changes to the management.yaml file.
Re-generate SDK and verify SDK generation is successfully.

a. Change directory to openapi. cd openapi

b. Run the build script to re-generate the Okta SDK Python package: generate.sh
Raise the PR mentioning details about the changes made to the management.yaml file.

if change isn't related to new openapi spec version, for example, update template for models, then:

edit needed templates under openapi/templates directory
re-generate okta-sdk-python (in openapi directory):

a. Change directory to openapi. cd openapi

b. Run the build script to re-generate the Okta SDK Python package: generate.sh

Tests

Execute pytest to run the tests.

Getting Started

Please follow the installation procedure and then run the following:

import asyncio
from okta import UserProfile, PasswordCredential, CreateUserRequest, UserNextLogin, UserCredentials
from okta.client import Client as OktaClient
config = {
    'orgUrl': 'https://{your_org}.okta.com',
    'token': 'YOUR_API_TOKEN',
}

okta_client = OktaClient(config)
user_config = {
    "firstName": "Sample",
    "lastName": "Sample",
    "email": "sample12.sample@example.com",
    "login": "sample12.sample@example.com",
    "mobilePhone": "555-415-1337"
  }
user_profile = UserProfile(**user_config)

password_value = {
    "value": "Knock*knock*neo*111"
}
password_credential = PasswordCredential(**password_value)
user_credentials = {
    "password": password_credential
}
user_credentials = UserCredentialsWritable(**user_credentials)
create_user_request = {
    "profile": user_profile,
    "credentials": user_credentials,
}
user_request = CreateUserRequest(**create_user_request)
async def users():
    next_login = UserNextLogin(UserNextLogin.CHANGEPASSWORD)
    user, resp, err = await okta_client.create_user(user_request, activate=True, provider=False, next_login=next_login)
    print("The response of UserApi->create_user:\n")
    print(user)
    print(resp, err)

    users, resp, err = await okta_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(users())

OAuth 2.0

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

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.

Extending the Client

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

Extending or Creating New Classes

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 code

and 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 code

Usage guide

These 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][users-api-docs] are organized under [the User api (okta.api.user_api.py)][users-api].

Asynchronous I/O is fairly new to Python after making its debut in Python 3.5. It's powered by the asyncio library which provides avenues to produce concurrent code. This allows developers to define async functions and await asynchronous calls within them. For more information, you can check out the [Python docs][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())

Authenticate a User

This library should only be used with the Okta management API. To call the [Authentication API][authn-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'})

Get and set custom headers

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

Exceptions

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

Pagination

This guide demonstrates how to use pagination with the Okta SDK for Python.

Overview

The Okta API returns paginated results for list operations (users, applications, groups, etc.). The SDK now includes built-in pagination utilities to make this easy!

How Okta Pagination Works

Okta uses cursor-based pagination with the after parameter. The Link response header contains the URL for the next page:

Link: <https://your-domain.okta.com/api/v1/users?after=CURSOR>; rel="next"

Quick Start - Using SDK Pagination Helpers (Recommended)

The SDK now provides helper functions that handle pagination automatically:

Method 1: `paginate_all()` - Iterate Through All Items

from okta.client import Client as OktaClient
from okta import paginate_all

async def fetch_all_users(okta_client):
    """Fetch all users automatically"""
    all_users = []

    # Automatically paginate through all users
    async for user in paginate_all(
        okta_client.list_users_with_http_info,
        limit=200  # Items per page
    ):
        all_users.append(user)
        print(f"User: {user.profile.email}")

    return all_users

Method 2: `paginate_pages()` - Process in Batches

from okta import paginate_pages

async def process_users_in_batches(okta_client):
    """Process users page by page"""

    async for page_data, page_num, has_more in paginate_pages(
        okta_client.list_users_with_http_info,
        limit=100,
        max_pages=5  # Optional: limit number of pages
    ):
        print(f"Processing page {page_num} with {len(page_data)} users")

        # Process the batch
        for user in page_data:
            process_user(user)

        if not has_more:
            print("Reached the last page")

Method 3: `PaginationHelper` - Manual Control

from okta import PaginationHelper

async def manual_pagination(okta_client):
    """Manually control pagination"""
    after_cursor = None

    while True:
        users, response, error = await okta_client.list_users_with_http_info(
            limit=200,
            after=after_cursor
        )

        if error or not users:
            break

        # Process users
        for user in users:
            print(user.profile.email)

        # Get next cursor using SDK helper
        after_cursor = PaginationHelper.extract_next_cursor(response.headers)

        if not after_cursor:
            break  # No more pages

SDK Pagination Utilities Reference

`paginate_all(api_method, limit, max_pages, **kwargs)`

Automatically paginate through all results, yielding individual items.

Parameters:

api_method: The _with_http_info method to call
limit: Number of results per page (default: 200)
max_pages: Maximum pages to fetch (default: None = unlimited)
**kwargs: Additional arguments for the API method

Yields: Individual items from all pages

Example:

# Fetch all applications with filtering
async for app in paginate_all(
    client.list_applications_with_http_info,
    limit=100,
    filter='status eq "ACTIVE"'
):
    print(app.label)

`paginate_pages(api_method, limit, max_pages, **kwargs)`

Paginate through results, yielding complete pages.

Parameters:

Same as paginate_all

Yields: Tuples of (page_data, page_number, has_more)

Example:

async for page, page_num, has_more in paginate_pages(
    client.list_groups_with_http_info,
    limit=50
):
    print(f"Page {page_num}: {len(page)} groups")
    # Process page...

`PaginationHelper.extract_next_cursor(headers)`

Extract the pagination cursor from response headers.

Parameters:

headers: HTTP response headers dictionary

Returns: The next page cursor string, or None if no more pages

Example:

users, response, error = await client.list_users_with_http_info(limit=100)
next_cursor = PaginationHelper.extract_next_cursor(response.headers)
if next_cursor:
    print(f"Next page available with cursor: {next_cursor}")

`PaginationHelper.has_next_page(headers)`

Check if more pages are available.

Returns: True if another page exists, False otherwise

`PaginationHelper.extract_pagination_info(headers)`

Extract all pagination information.

Returns: Dict with {'next': cursor, 'prev': cursor, 'self': cursor}

Manual Implementation Example

If you need more control, you can implement pagination manually:

Basic Pagination Pattern

async def paginate_users(okta_client):
    """Fetch all users using pagination"""
    all_users = []
    after_cursor = None

    while True:
        # Use _with_http_info to get response headers
        users, response, error = await okta_client.list_users_with_http_info(
            limit=200,  # Maximum per page
            after=after_cursor
        )

        if error or not users:
            break

        all_users.extend(users)

        # Extract next cursor from Link header
        after_cursor = extract_next_cursor(response.headers)

        if not after_cursor:
            break  # No more pages

    return all_users

Extracting the Next Cursor

def extract_next_cursor(headers):
    """Extract the 'after' cursor from Link header"""
    link_header = headers.get('Link') or headers.get('link')

    if not link_header:
        return None

    # Parse Link header: <URL>; rel="next"
    links = link_header.split(',')
    for link in links:
        if 'rel="next"' in link:
            url = link.split(';')[0].strip('<>').strip()
            if 'after=' in url:
                after_param = url.split('after=')[1]
                return after_param.split('&')[0]

    return None

Complete Example

See comprehensive_okta_test.py for a complete working example. Key methods:

list_users_with_pagination() - Demonstrates user pagination (Step 7b)
list_applications_with_pagination() - Demonstrates application pagination (Step 7c)
_extract_next_cursor() - Helper method to parse Link headers

Running the Example

# Install dependencies
pip install okta-sdk-python

# Update credentials in comprehensive_okta_test.py
# Then run:
python comprehensive_okta_test.py

Key Points

Use _with_http_info methods: These return (data, response, error) tuples that include headers
Parse the Link header: Extract the after cursor for the next page
Set appropriate limits: Maximum is usually 200 per page
Handle rate limits: Okta has rate limits, so implement backoff strategies for production

Available Paginated Methods

Most list operations support pagination:

list_users_with_http_info()
list_applications_with_http_info()
list_groups_with_http_info()
list_policies_with_http_info()
And many more...

Complete Working Examples

Example 1: Fetch All Users (Simple)

import asyncio
from okta.client import Client as OktaClient
from okta import paginate_all

async def main():
    config = {
        'orgUrl': 'https://your-domain.okta.com',
        'token': 'your-api-token',
    }

    client = OktaClient(config)

    # Collect all users
    all_users = [user async for user in paginate_all(
        client.list_users_with_http_info,
        limit=200
    )]

    print(f"Total users: {len(all_users)}")

if __name__ == "__main__":
    asyncio.run(main())

Example 2: Process Users in Batches (Memory Efficient)

import asyncio
from okta.client import Client as OktaClient
from okta import paginate_pages

async def process_users_in_batches():
    config = {
        'orgUrl': 'https://your-domain.okta.com',
        'token': 'your-api-token',
    }

    client = OktaClient(config)
    total_processed = 0

    async for page_data, page_num, has_more in paginate_pages(
        client.list_users_with_http_info,
        limit=200
    ):
        # Process this batch
        print(f"Processing page {page_num}: {len(page_data)} users")

        for user in page_data:
            # Do something with each user
            print(f"  - {user.profile.email}")
            total_processed += 1

        # Optional: add delay for rate limiting
        if has_more:
            await asyncio.sleep(0.5)

    print(f"Total processed: {total_processed} users")

if __name__ == "__main__":
    asyncio.run(process_users_in_batches())

Example 3: Filtered Pagination with Error Handling

import asyncio
from okta.client import Client as OktaClient
from okta import paginate_all

async def fetch_active_applications():
    config = {
        'orgUrl': 'https://your-domain.okta.com',
        'token': 'your-api-token',
    }

    client = OktaClient(config)
    active_apps = []

    try:
        async for app in paginate_all(
            client.list_applications_with_http_info,
            limit=100,
            filter='status eq "ACTIVE"',
            max_pages=10  # Safety limit
        ):
            active_apps.append(app)
            print(f"Found: {app.label}")

    except Exception as e:
        print(f"Error during pagination: {e}")

    print(f"Total active applications: {len(active_apps)}")
    return active_apps

if __name__ == "__main__":
    asyncio.run(fetch_active_applications())

Example 4: Manual Pagination with SDK Helper

import asyncio
from okta.client import Client as OktaClient
from okta import PaginationHelper

async def manual_pagination_example():
    config = {
        'orgUrl': 'https://your-domain.okta.com',
        'token': 'your-api-token',
    }

    client = OktaClient(config)

    after_cursor = None
    page_count = 0
    total_groups = 0

    while True:
        # Fetch page
        groups, response, error = await client.list_groups_with_http_info(
            limit=50,
            after=after_cursor
        )

        if error:
            print(f"Error: {error}")
            break

        if not groups:
            break

        page_count += 1
        total_groups += len(groups)
        print(f"Page {page_count}: {len(groups)} groups")

        # Check for next page using SDK helper
        if not PaginationHelper.has_next_page(response.headers):
            print("Reached last page")
            break

        # Get next cursor
        after_cursor = PaginationHelper.extract_next_cursor(response.headers)

        # Rate limiting
        await asyncio.sleep(0.5)

    print(f"Total: {total_groups} groups across {page_count} pages")

if __name__ == "__main__":
    asyncio.run(manual_pagination_example())

Best Practices

Use SDK helpers: Prefer paginate_all() or paginate_pages() for cleaner code
Set appropriate limits: Use 100-200 per page in production (don't use 5 like in demos)
Handle rate limits: Add delays between pages or implement exponential backoff
Process incrementally: For large datasets, use paginate_pages() to avoid memory issues
Set safety limits: Use max_pages parameter to prevent runaway pagination
Handle errors: Always wrap pagination in try-except blocks
Use filtering: Apply filters to reduce the amount of data fetched

Additional Resources

Logging

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!

Configuration reference

This library looks for configuration in the following sources:

An okta.yaml file in a .okta folder in the current user's home directory (~/.okta/okta.yaml or %userprofile%\.okta\okta.yaml). See a sample YAML Configuration
A okta.yaml file 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.

YAML configuration

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: INFO

When 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: INFO

If a proxy is not going to be used for the SDK, you may omit the okta.client.proxy section from your okta.yaml file

Environment variables

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_AUTHORIZATIONMODE
OKTA_CLIENT_ORGURL
OKTA_CLIENT_TOKEN
OKTA_CLIENT_CLIENTID
OKTA_CLIENT_SCOPES
OKTA_CLIENT_PRIVATEKEY
OKTA_CLIENT_USERAGENT
OKTA_CLIENT_CONNECTIONTIMEOUT
OKTA_CLIENT_REQUESTTIMEOUT
OKTA_CLIENT_CACHE_ENABLED
OKTA_CLIENT_CACHE_DEFAULTTTI
OKTA_CLIENT_CACHE_DEFAULTTTL
OKTA_CLIENT_PROXY_PORT
OKTA_CLIENT_PROXY_HOST
OKTA_CLIENT_PROXY_USERNAME
OKTA_CLIENT_PROXY_PASSWORD
OKTA_CLIENT_RATELIMIT_MAXRETRIES
OKTA_TESTING_TESTINGDISABLEHTTPSCHECK

Other configuration options

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

Rate Limiting

The Okta API will return 429 responses if too many requests are made within a given time. Please see [Rate Limiting at Okta][rate-limiting-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.

Built-In Retry

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.

Building the SDK

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 build from the root of the project (assuming Python is installed)
Ensure tests run succesfully. Install tox if not installed already using: pip install tox. Run tests using tox in the root directory of the project.

Contributing

We're happy to accept contributions and PRs! Please see the Contribution Guide to understand how to structure a contribution.

For contributing, please make changes to the mustache templates and submit the change.

Doc Guide

Please find the link to API endpoints and Models doc guide here: Doc Guide

Author

devex-public@okta.com

Requirements.

Release status

Need help?

Installation & Usage

pip install

Setuptools

Generate models and APIs

Prerequisites

if change is needed because of new version of openapi spec that has been released, then:

if change isn't related to new openapi spec version, for example, update template for models, then:

Tests

Getting Started

OAuth 2.0

Extending the Client

Extending or Creating New Classes

Usage guide

Authenticate a User

Get and set custom headers

Exceptions

Pagination

Overview

How Okta Pagination Works

Quick Start - Using SDK Pagination Helpers (Recommended)

Method 1: paginate_all() - Iterate Through All Items

Method 2: paginate_pages() - Process in Batches

Method 3: PaginationHelper - Manual Control

SDK Pagination Utilities Reference

paginate_all(api_method, limit, max_pages, **kwargs)

paginate_pages(api_method, limit, max_pages, **kwargs)

PaginationHelper.extract_next_cursor(headers)

PaginationHelper.has_next_page(headers)

PaginationHelper.extract_pagination_info(headers)

Manual Implementation Example

Basic Pagination Pattern

Extracting the Next Cursor

Complete Example

Running the Example

Key Points

Available Paginated Methods

Complete Working Examples

Example 1: Fetch All Users (Simple)

Example 2: Process Users in Batches (Memory Efficient)

Example 3: Filtered Pagination with Error Handling

Example 4: Manual Pagination with SDK Helper

Best Practices

Additional Resources

Logging

Configuration reference

YAML configuration

Environment variables

Other configuration options

Rate Limiting

Built-In Retry

Building the SDK

Contributing

Doc Guide

Author

Method 1: `paginate_all()` - Iterate Through All Items

Method 2: `paginate_pages()` - Process in Batches

Method 3: `PaginationHelper` - Manual Control

`paginate_all(api_method, limit, max_pages, **kwargs)`

`paginate_pages(api_method, limit, max_pages, **kwargs)`

`PaginationHelper.extract_next_cursor(headers)`

`PaginationHelper.has_next_page(headers)`

`PaginationHelper.extract_pagination_info(headers)`