user_context module

Management of user contexts for the bot. User contexts are either stored in Redis or in a local file.

TokenManager: abstract base class for token managers
YAMLTokenManager: token manager storing tokens for users in local YAML file
RedisTokenManager: token manager storing tokens and OAuth flow state in Redis

class user_context.UserContext(*, user_id: str, tokens: Tokens)[source]

Bases: BaseModel

User context. For each user we need to keep track of the tokens obtained for that user

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

user_id: str: user id

tokens: Tokens: tokens for the user: wxc_sdk.tokens.Tokens

class user_context.TokenManager(bot_token: str, integration: Integration, **kwargs)[source]

Bases: ABC

Token manager consolidates all functions around tokens.

registering a new OAuth flow
processing the code at the end of the flow to obtain tokens
storing user contexts
providing access to user contexts

Parameters:

bot_token – bot Token
integration – wxc_sdk.integration.Integration
kwargs –

abstract close()[source]

abstract start_flow(*, user_id: str) → str[source]

Register OAuth flow for a user

Parameters:: user_id (str) – user id to register a flow for
Returns:: flow id for the new flow
Return type:: str

abstract process_redirect(*, flow_id: str, code: str) → str[source]

Process redirect at end of OAuth flow. New tokens are stored in user context

Parameters:

flow_id (str) – OAuth flow id
code (str) – code obtained from final URL in OAuth flow

Returns:

text for HTTP response

Return type:

str

abstract get_user_context(*, user_id: str) → UserContext | None[source]

Get user context for given user_id

Parameters:: user_id (str) – id of user to get context for
Returns:: user context
Return type:: UserContext

set_user_context(*, user_id: str, user_context: UserContext | None = None)[source]

Add user context to cache or clear it from cache (user_context==None)

Parameters:

user_id (str) – user id
user_context (UserContext) – user context to set; if None then clear user context for this user

register_redirect(*, flask: Flask)[source]

Register /redirect endpoint for OAuth flows.

redirect() is registered as view handler with flask by calling flask.Flask.add_url_rule(). The view handler is called whenever a request on the /redirect endpoint needs to be handled.

Parameters:: flask (flask.Flask) – Flask app to register the /redirect endpoint with

redirect()[source]

view function for GET on /redirect at end of authorization flow.

Get code and state (flow id) from URL and set event on the registered flow

The view function is called whenever a request on /redirect needs to be handled. Request details are available in the global flask.request variable (imported as flask_request).

Authorization code and flow id are contained in the query string of the request URL. The parsed arguments of the query string can be found in the flask.Request.args attribute of flask_request.

token_refresh(*, tokens: Tokens) → bool[source]

try to refresh the tokens

Parameters:: tokens (wxc_sdk.tokens.Tokens) – tokens to refresh
Returns:: True -> tokens got updated

get_user_context_and_refresh(*, user_id: str) → UserContext | None[source]

Get user context for given user_id and try to refresh the access token if needed

Parameters:: user_id (str) – id of user to get context for
Returns:: user context
Return type:: UserContext

class user_context.YAMLTokenManager(bot_token: str, integration: Integration, yml_base: str)[source]

Bases: TokenManager

A TokenManager using a local YAML file to store user contexts. Flow state is stored within the instance. This only works if there is a single worker servicing the requests.

Parameters:

bot_token (str) – Bot token
integration (wxc_sdk.integration.Integration) – Integration to use to obtain tokens
yml_base (str) – base name (no extension) for YAML file to use to persist state

close()[source]

start_flow(*, user_id: str) → str[source]

Register OAuth flow for a user

Parameters:: user_id –
Returns:: flow id

process_redirect(*, flow_id: str, code: str) → str[source]

Process redirect at end of OAuth flow. New tokens are stored in user context

Parameters:

flow_id – OAuth flow id
code – code obtained from final URL in OAuth flow

Returns:

text for HTTP response

set_user_context(*, user_id: str, user_context: UserContext | None = None)[source]

Store user context in redis

Parameters:

user_id (str) – user id
user_context (UserContext) – user context; if None then clear the user context

get_user_context(*, user_id: str) → UserContext | None[source]

Get user context for given user_id

Parameters:: user_id (str) –
Returns:
Return type:: UserContext

class user_context.RedisTokenManager(bot_token: str, integration: Integration, redis_host: str | None = None, redis_url: str | None = None)[source]

Bases: TokenManager

Token Manager using Redis as backend

set up Redis token Manager

Parameters:

bot_token (str) – bot access token. Required to be able to send responses to the user
integration (:class:wxc_sdk.integration.Integration) – OAuth integration. Used to call the respective endpoints to obtain/refresh tokens
redis_host (str) – Redis host, Redis host takes precedence over Redis URL
redis_url (str) – Redis URL, Redis host takes precedence over Redis URL

FLOW_SET = 'flows': Redis key for set of active flow ids

FLOW_MAINTENANCE = 'flow-maintenance': Redis key for datetime of latest flow maintenance. This is used in flow_maintenance_needed() to determine whether flow maintenance is needed

USER_KEY_PREFIX = 'user': prefix for Redis keys to store UserContext for users. Full key is in the form of ‘{USER_KEY_PREFIX}-{user_id}’

USER_SET = 'users': Redis key for set of users IDs we have UserContext for in Redis

class FlowState(*, user_id: str, created: datetime = None)[source]

Bases: BaseModel

keep track of a pending OAuth flow. For each flow we keep track of the creation time. This time is used to garbage collect old flows if needed

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

user_id: str: user id the flow was created for

created: datetime: datetime.datetime the flow was created at

flow_maintenance_needed(*, force: bool = False) → bool[source]

Determine whether flow maintenance is needed; only once every 15 minutes

Returns:

flow_maintenance(*, force: bool = False, output: TextIOBase | None = None)[source]

Garbage collection on existing flows if needed.

Returns:

start_flow(*, user_id: str) → str[source]

Register OAuth flow for a user

Parameters:: user_id (str) – id of user to start the flow for
Returns:: flow id, the flow id is used as state parameter in OAuth flow and allows to identify the flow when acting on the POST at the end of the OAuth flow

process_redirect(*, flow_id: str, code: str) → str[source]

Process redirect at end of OAuth flow. New tokens are stored in user context

Parameters:

flow_id (str) – OAuth flow id
code (str) – code obtained from final URL in OAuth flow

Returns:

text for HTTP response

Return type:

str

set_user_context(*, user_id: str, user_context: UserContext | None = None)[source]

Store user context in redis

Parameters:

user_id (str) – user id of user to store context for
user_context (UserContext) – context to store; if None then clear context for this user

get_user_context(*, user_id: str) → UserContext | None[source]

Get user context for given user_id

Parameters:: user_id (str) –
Returns:
Return type:: UserContext