Database context

class luna_accounts.db.context.DBContext(logger)[source]

Accounts DB context.

async createAccount(account, accountId=None)[source]
Return type:

str

Create new account :param account: account to create :param accountId: account id to create

Returns:

unique account id

Raises:

VLException(Error.AccountAlreadyExist.format(account.login), 409, False) if account with the same login – already exists

async createToken(token, accountId)[source]
Return type:

tuple[str, str]

Create new token :param token: token to create :param accountId: account id

Returns:

tuple with unique token and its’ id

Raises:

VLException(Error.AccountNotFound.format(accountId), 400, False) if specified account id not found

async deleteAccount(accountId)[source]
Return type:

bool

Delete account by account id :param accountId: id of account

Returns:

True if account was deleted otherwise False

async deleteToken(tokenId, accountId)[source]
Return type:

bool

Delete token by id :param tokenId: token id :param accountId: account id

Returns:

True if token was removed, False if token was not found

async getAccounts(targets, page=1, pageSize=1, login=None, accountType=None, createTimeLt=None, createTimeGte=None, accountId=None, getCount=True)[source]
Return type:

tuple[list[dict], int] | list[dict]

Get accounts with pagination :param targets: targets to get :param page: page :param pageSize: page size :param login: account login :param accountType: account type :param createTimeLt: upper bound of account create time :param createTimeGte: lower bound of account create time :param accountId: account id :param getCount: whether to get account count

Returns:

accounts or accounts and count

async getAccountsCount()[source]
Return type:

int

Get accounts count :returns: accounts count

async getToken(tokenId, permissionTargets=0, accountId=None)[source]
Return type:

dict | None

Get token :param tokenId: token id :param permissionTargets: permission targets :param accountId: account id

Returns:

token if exist otherwise None

async getTokens(page, pageSize, permissionTargets=None, accountId=None, createTimeLt=None, createTimeGte=None)[source]
Return type:

list[dict]

Get tokens :param accountId: account id :param page: page :param pageSize: page size :param permissionTargets: permission targets :param createTimeLt: upper bound of token create time :param createTimeGte: lower bound of token create time

Returns:

tokens list

async classmethod initDBContext(dbSettings, storageTime, ecdsaKeyString=None, ecdsaKeyPassword=None, **kwargs)[source]
Return type:

None

Initialize context :param dbSettings: database settings :param storageTime: storage time :param ecdsaKeyString: rsa key string for RS256 :param ecdsaKeyPassword: rsa key password for RS256

classmethod makeOutputAccounts(rows, columns)[source]

Make result accounts (from the database reply) proper for user.

Return type:

list[dict[str, any]]

Parameters:
  • rows – list from db

  • columns – selected columns

Returns:

faces with changed fields

makeOutputToken(rowFromDB, permissionTargets=None)[source]

Make dict with token from row from db

Return type:

dict

async patchAccount(accountId, accountOverride)[source]
Return type:

bool

Patch account by account id :param accountId: id of account :param accountOverride: account for patch model

Returns:

True if account was updated otherwise False

async replaceToken(tokenId, token, accountId=None)[source]
Return type:

str

Replace existing token using specified token id :param tokenId: token id :param accountId: account id :param token: token

Returns:

token

Raises:

VLException(Error.TokenNotFoundById.format(tokenId), 404, False) if token not found

async verifyAccountByAccountId(accountId)[source]
Return type:

str

Verify account by account id :param accountId: accountId

Returns:

account type if account found otherwise None

async verifyAccountByLoginPassword(login, password)[source]
Return type:

tuple[str, str]

Verify account by login and password :param login: login :param password: password

Returns:

account type and account id

Raises:

VLException(Error.AccountLoginPasswordIncorrect, 400, isCriticalError=False) if account not found or – password doesn’t match

async verifyToken(token, permissionTargets=None)[source]
Return type:

Optional[tuple[str, dict]]

Verify account by JWT token :param token: JWT token :param permissionTargets: target for permissions to return in response

Returns:

tuple with account type and permissions if account found otherwise None

Raises:
  • VLException(Error.JWTTokenNotFound, 400, isCriticalError=False) if token not found or unexpected format

  • VLException(Error.TokenExpired, 400, False) if token expired