Client

Enhanced client library with connection pooling, retry logic, and circuit breaker

class auth.client.RetryableHTTPAdapter(retries=3, backoff_factor=0.3, status_forcelist=(500, 502, 504), **kwargs)

Bases: HTTPAdapter

HTTP adapter with retry logic

__init__(retries=3, backoff_factor=0.3, status_forcelist=(500, 502, 504), **kwargs)

init_poolmanager(*args, **kwargs)

Initializes a urllib3 PoolManager.

This method should not be called from user code, and is only exposed for use when subclassing the HTTPAdapter.

Parameters:

• connections – The number of urllib3 connection pools to cache.

• maxsize – The maximum number of connections to save in the pool.

• block – Block when no free connections are available.

• pool_kwargs – Extra keyword arguments used to initialize the Pool Manager.

class auth.client.EnhancedAuthClient(api_key: str, service_url: str, max_retries: int = 3, pool_connections: int = 10, pool_maxsize: int = 20, timeout: int = 30, circuit_breaker_enabled: bool = True)

Bases: object

Enhanced client with connection pooling, retry logic, and circuit breaker

Initialize the enhanced client

Parameters:

• api_key – The API key for authentication

• service_url – The base URL for the auth service

• max_retries – Number of times to retry failed requests

• pool_connections – Number of connection pools

• pool_maxsize – Max connections per pool

• timeout – Request timeout in seconds

• circuit_breaker_enabled – Whether circuit breaker is enabled

__init__(api_key: str, service_url: str, max_retries: int = 3, pool_connections: int = 10, pool_maxsize: int = 20, timeout: int = 30, circuit_breaker_enabled: bool = True)

Initialize the enhanced client

Parameters:

• api_key – The API key for authentication

• service_url – The base URL for the auth service

• max_retries – Number of times to retry failed requests

• pool_connections – Number of connection pools

• pool_maxsize – Max connections per pool

• timeout – Request timeout in seconds

• circuit_breaker_enabled – Whether circuit breaker is enabled

ping() → Dict[str, Any]

Health check

add_membership(user: str, group: str) → Dict[str, Any]

Add user to a group

remove_membership(user: str, group: str) → Dict[str, Any]

Remove user from a group

has_membership(user: str, group: str) → Dict[str, Any]

Check if user is member of a group

add_permission(group: str, name: str) → Dict[str, Any]

Add permission to a group

remove_permission(group: str, name: str) → Dict[str, Any]

Remove permission from a group

has_permission(group: str, name: str) → Dict[str, Any]

Check if group has permission

user_has_permission(user: str, name: str) → Dict[str, Any]

Check if user has permission

get_user_permissions(user: str) → Dict[str, Any]

Get all permissions for a user

get_role_permissions(role: str) → Dict[str, Any]

Get all permissions for a role

get_user_roles(user: str) → Dict[str, Any]

Get all roles for a user

get_role_members(role: str) → Dict[str, Any]

Get all members of a role

list_roles() → Dict[str, Any]

List all roles

which_roles_can(name: str) → Dict[str, Any]

Get roles that can perform an action

which_users_can(name: str) → Dict[str, Any]

Get users that can perform an action

create_role(role: str) → Dict[str, Any]

Create a new role

delete_role(role: str) → Dict[str, Any]

Delete a role

get_users_for_workflow(workflow_name: str) → Dict[str, Any]

Get all users who can run a specific workflow

check_user_workflow_permission(user: str, workflow_name: str) → Dict[str, Any]

Check if a user can run a specific workflow

close()

Close the session

class auth.client.Client(api_key: str, service_url: str, max_retries: int = 3, pool_connections: int = 10, pool_maxsize: int = 20, timeout: int = 30, circuit_breaker_enabled: bool = True)

Bases: EnhancedAuthClient

Legacy client class for backward compatibility

Initialize the enhanced client

Parameters:

• api_key – The API key for authentication

• service_url – The base URL for the auth service

• max_retries – Number of times to retry failed requests

• pool_connections – Number of connection pools

• pool_maxsize – Max connections per pool

• timeout – Request timeout in seconds

• circuit_breaker_enabled – Whether circuit breaker is enabled

EnhancedAuthClient

class auth.client.EnhancedAuthClient(api_key: str, service_url: str, max_retries: int = 3, pool_connections: int = 10, pool_maxsize: int = 20, timeout: int = 30, circuit_breaker_enabled: bool = True)

Bases: object

Enhanced client with connection pooling, retry logic, and circuit breaker

Initialize the enhanced client

Parameters:

• api_key – The API key for authentication

• service_url – The base URL for the auth service

• max_retries – Number of times to retry failed requests

• pool_connections – Number of connection pools

• pool_maxsize – Max connections per pool

• timeout – Request timeout in seconds

• circuit_breaker_enabled – Whether circuit breaker is enabled

__init__(api_key: str, service_url: str, max_retries: int = 3, pool_connections: int = 10, pool_maxsize: int = 20, timeout: int = 30, circuit_breaker_enabled: bool = True)

Initialize the enhanced client

Parameters:

• api_key – The API key for authentication

• service_url – The base URL for the auth service

• max_retries – Number of times to retry failed requests

• pool_connections – Number of connection pools

• pool_maxsize – Max connections per pool

• timeout – Request timeout in seconds

• circuit_breaker_enabled – Whether circuit breaker is enabled

ping() → Dict[str, Any]

Health check

add_membership(user: str, group: str) → Dict[str, Any]

Add user to a group

remove_membership(user: str, group: str) → Dict[str, Any]

Remove user from a group

has_membership(user: str, group: str) → Dict[str, Any]

Check if user is member of a group

add_permission(group: str, name: str) → Dict[str, Any]

Add permission to a group

remove_permission(group: str, name: str) → Dict[str, Any]

Remove permission from a group

has_permission(group: str, name: str) → Dict[str, Any]

Check if group has permission

user_has_permission(user: str, name: str) → Dict[str, Any]

Check if user has permission

get_user_permissions(user: str) → Dict[str, Any]

Get all permissions for a user

get_role_permissions(role: str) → Dict[str, Any]

Get all permissions for a role

get_user_roles(user: str) → Dict[str, Any]

Get all roles for a user

get_role_members(role: str) → Dict[str, Any]

Get all members of a role

list_roles() → Dict[str, Any]

List all roles

which_roles_can(name: str) → Dict[str, Any]

Get roles that can perform an action

which_users_can(name: str) → Dict[str, Any]

Get users that can perform an action

create_role(role: str) → Dict[str, Any]

Create a new role

delete_role(role: str) → Dict[str, Any]

Delete a role

get_users_for_workflow(workflow_name: str) → Dict[str, Any]

Get all users who can run a specific workflow

check_user_workflow_permission(user: str, workflow_name: str) → Dict[str, Any]

Check if a user can run a specific workflow

close()

Close the session

Client (Legacy)

class auth.client.Client(api_key: str, service_url: str, max_retries: int = 3, pool_connections: int = 10, pool_maxsize: int = 20, timeout: int = 30, circuit_breaker_enabled: bool = True)

Bases: EnhancedAuthClient

Legacy client class for backward compatibility

Initialize the enhanced client

Parameters:

• api_key – The API key for authentication

• service_url – The base URL for the auth service

• max_retries – Number of times to retry failed requests

• pool_connections – Number of connection pools

• pool_maxsize – Max connections per pool

• timeout – Request timeout in seconds

• circuit_breaker_enabled – Whether circuit breaker is enabled