TokenDB

Token class is a front-end to the TokenDB Database. Long-term user tokens are stored here, which can be used to obtain new tokens.

class DIRAC.FrameworkSystem.DB.TokenDB.Token(**kwargs)

Bases: sqlalchemy.orm.decl_api.Base, authlib.integrations.sqla_oauth2.tokens_mixins.OAuth2TokenMixin

This class describe token fields

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

access_token

access_token_revoked_at

check_client(client)

A method to check if this token is issued to the given client. For instance, client_id is saved on token table:

def check_client(self, client):
    return self.client_id == client.client_id

Returns

bool

client_id

expires_at

expires_in

get_expires_in()

A method to get the expires_in value of the token. e.g. the column is called expires_in:

def get_expires_in(self):
    return self.expires_in

Returns

timestamp int

get_scope()

A method to get scope of the authorization code. For instance, the column is called scope:

def get_scope(self):
    return self.scope

Returns

scope string

id

is_expired()

A method to define if this token is expired. For instance, there is a column expired_at in the table:

def is_expired(self):
    return self.expired_at < now

Returns

boolean

is_revoked()

A method to define if this token is revoked. For instance, there is a boolean column revoked in the table:

def is_revoked(self):
    return self.revoked

Returns

boolean

issued_at

kid

metadata = MetaData()

provider

refresh_token

refresh_token_revoked_at

registry = <sqlalchemy.orm.decl_api.registry object>

rt_expires_at

scope

token_type

user_id

class DIRAC.FrameworkSystem.DB.TokenDB.TokenDB(*args, **kwargs)

Bases: DIRAC.Core.Base.SQLAlchemyDB.SQLAlchemyDB

TokenDB class is a front-end to the TokenDB Database

__init__(*args, **kwargs)

Constructor

delete(table, params)

Parameters

• table (str) – table from where to delete

• params (dict) – dictionary of which line(s) to delete

Returns

S_OK() || S_ERROR()

getCSOption(optionName, defaultValue=None)

getTokenForUserProvider(userID, provider)

Get token for user ID and identity provider name

Parameters

• userID (str) – user ID

• provider (str) – provider name

Returns

S_OK(OAuth2Token)/S_ERROR() – return an OAuth2Token object, which is also a dict

getTokensByUserID(userID)

Return tokens for user ID

Parameters

userID (str) – user ID that return identity provider

Returns

S_OK(list)/S_ERROR() – tokens as OAuth2Token objects

insert(table, params)

Inserts params in the DB.

Parameters

• table (str) – table where to insert

• params (dict) – Dictionary to fill a single line

Returns

S_OK() || S_ERROR()

removeToken(access_token=None, refresh_token=None, user_id=None)

Remove token from DB

Parameters

• access_token (str) – access token

• refresh_token (str) – refresh token

Returns

S_OK(str)/S_ERROR()

select(table, params)

Uses params to build conditional SQL statement ( WHERE … ).

Parameters

params - dict

arguments for the mysql query ( must match table columns ! ).

Returns

S_OK() || S_ERROR()

updateToken(token: dict, userID: str, provider: str, rt_expired_in: int)

Store or update an existing token in the database. Before saving, the token is checked for expiration. Also, the database cannot contain several user tokens signed by one provider, only one with the maximum possible permissions is enough.

Parameters

• token – token information dictionary

• userID – user ID (token owner)

• provider – provider name that issued the token

• rt_expired_in – refresh token expiration time, will be applied if the rt_expires_at value is missing

Returns

S_OK(list)/S_ERROR() – return old tokens that should be revoked.