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 = MetaData()
provider
refresh_token
refresh_token_revoked_at
registry: RegistryType = <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.