Skip to content

jam.aio.instance

instance

Classes:

Name Description
Jam

Main async Jam instance.

Jam

Jam(
    config: str | dict[str, Any] = "pyproject.toml",
    pointer: str = "jam",
    *,
    logger: type[BaseLogger] = JamLogger,
    log_level: Literal[
        "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"
    ] = "INFO",
    serializer: BaseEncoder
    | type[BaseEncoder] = JsonEncoder,
    plugins: list[type[BasePlugin]] = [],
)

Bases: BaseAsyncJam

Main async Jam instance.

Methods:

Name Description
jwe_decrypt

Decrypt JWE token.

jwe_encrypt

Encrypt data using JWE.

jws_sign

Sign data using JWS.

jws_verify

Verify JWS token.

jwt_create

Create JWT token.

jwt_decode

Verify and decode JWT token.

jwt_encode

Encode the JWT with the given expire, header, and payload.

jwt_make_payload

Make JWT-specific payload.

oauth2_client_credentials_flow

Obtain access token using client credentials flow (no user interaction).

oauth2_fetch_token

Exchange authorization code for access token.

oauth2_get_authorized_url

Generate full OAuth2 authorization URL.

oauth2_refresh_token

Use refresh token to obtain a new access token.

otp_code

Generates an OTP.

otp_uri

Generates an otpauth:// URI for Google Authenticator.

otp_verify_code

Checks the OTP code, taking into account the acceptable window.

paseto_create

Create new PASETO.

paseto_decode

Decode PASETO.

paseto_make_payload

Generate payload for PASETO.

session_clear

Delete all sessions by key.

session_create

Create new session.

session_delete

Delete session.

session_get

Get data from session.

session_rework

Rework session.

session_update

Update session data.

jwe_decrypt async

jwe_decrypt(token: str) -> bytes

Decrypt JWE token.

Parameters:

Name Type Description Default
token str

JWE token.

required

Returns:

Name Type Description
bytes bytes

Decrypted data.

jwe_encrypt async

jwe_encrypt(
    data: dict[str, Any] | str,
    header: dict[str, Any] | None = None,
) -> str

Encrypt data using JWE.

Parameters:

Name Type Description Default
data dict[str, Any] | str

Data to encrypt. If dict, will be JSON encoded.

required
header dict[str, Any] | None

JWE header.

None

Returns:

Name Type Description
str str

JWE token.

jws_sign async

jws_sign(
    data: dict[str, Any] | str,
    header: dict[str, Any] | None = None,
) -> str

Sign data using JWS.

Parameters:

Name Type Description Default
data dict[str, Any] | str

Data to sign. If dict, will be JSON encoded.

required
header dict[str, Any] | None

JWS header.

None

Returns:

Name Type Description
str str

JWS token.

jws_verify async

jws_verify(token: str) -> dict[str, Any]

Verify JWS token.

Parameters:

Name Type Description Default
token str

JWS token.

required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: Decoded payload.

jwt_create async

jwt_create(payload: dict[str, Any]) -> str

Create JWT token.

Deprecated

Use Jam.jwt_encode

Parameters:

Name Type Description Default
payload dict[str, Any]

Data payload

required

Returns:

Name Type Description
str str

New token

jwt_decode async

jwt_decode(
    token: str,
    check_exp: bool = True,
    check_list: bool = True,
    check_nbf: bool = False,
    include_headers: bool = False,
) -> dict[str, Any]

Verify and decode JWT token.

Parameters:

Name Type Description Default
token str

JWT token

required
check_exp bool

Check expire

True
check_list bool

Check white/black list. Docs: https://jam.makridenko.ru/jwt/lists/what/

True
check_nbf bool

Check not-before time

False
include_headers bool

Include headers in the decoded payload

False

Returns:

Type Description
dict[str, Any]

dict[str, Any]: Decoded payload

Raises:

Type Description
JamJWTExpired

If token is expired

JamJWTNotYetValid

If token is not yet valid (nbf claim)

JamConfigurationError

If JWT list is not connected

JamJWTNotInWhiteList

If token is not in white list

JamJWTInBlackList

If token is in black list

jwt_encode async

jwt_encode(
    iss: str | None = None,
    sub: str | None = None,
    aud: str | None = None,
    exp: int | None = None,
    nbf: int | None = None,
    jti: str | None = None,
    *,
    payload: dict[str, Any] | None = None,
    header: dict[str, Any] | None = None,
) -> str

Encode the JWT with the given expire, header, and payload.

Parameters:

Name Type Description Default
exp int | None

The expiration time in seconds.

None
nbf int | None

The not-before time in seconds.

None
iss str | None

The issuer.

None
sub str | None

The subject.

None
aud str | None

The audience.

None
jti str | None

The JWT ID. If none use the JTI fabric function.

None
header dict[str, Any] | None

The header to include in the JWT.

None
payload dict[str, Any] | None

The payload to include in the JWT.

None

Returns:

Name Type Description
str str

The encoded JWT.

jwt_make_payload async

jwt_make_payload(
    exp: int | None, data: dict[str, Any]
) -> dict[str, Any]

Make JWT-specific payload.

Deprecated

This method is deprecated; the JWT payload is generated automatically in accordance with the specification.

Parameters:

Name Type Description Default
exp int | None

Token expire

required
data dict[str, Any]

Data to payload

required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: Payload

oauth2_client_credentials_flow async

oauth2_client_credentials_flow(
    provider: str,
    scope: list[str] | None = None,
    **extra_params: Any,
) -> dict[str, Any]

Obtain access token using client credentials flow (no user interaction).

Parameters:

Name Type Description Default
provider str

OAuth2 provider

required
scope list[str] | None

Auth scope

None
extra_params Any

Extra auth params if needed

{}

Raises:

Type Description
JamOAuth2EmptyRaw

If response is empty

JamOAuth2Error

HTTP error

Returns:

Name Type Description
dict dict[str, Any]

JSON with access token

oauth2_fetch_token async

oauth2_fetch_token(
    provider: str,
    code: str,
    grant_type: str = "authorization_code",
    **extra_params: Any,
) -> dict[str, Any]

Exchange authorization code for access token.

Parameters:

Name Type Description Default
provider str

Provider name

required
code str

OAuth2 code

required
grant_type str

Type of oauth2 grant

'authorization_code'
extra_params Any

Extra auth params if needed

{}

Returns:

Name Type Description
dict dict[str, Any]

OAuth2 token

oauth2_get_authorized_url async

oauth2_get_authorized_url(
    provider: str, scope: list[str], **extra_params: Any
) -> str

Generate full OAuth2 authorization URL.

Parameters:

Name Type Description Default
provider str

Provider name

required
scope list[str]

Auth scope

required
extra_params Any

Extra ath params

{}

Returns:

Name Type Description
str str

Authorization url

oauth2_refresh_token async

oauth2_refresh_token(
    provider: str,
    refresh_token: str,
    grant_type: str = "refresh_token",
    **extra_params: Any,
) -> dict[str, Any]

Use refresh token to obtain a new access token.

Parameters:

Name Type Description Default
provider str

Provider name

required
refresh_token str

Refresh token

required
grant_type str

Grant type

'refresh_token'
extra_params Any

Extra auth params if needed

{}

Returns:

Name Type Description
dict dict[str, Any]

Refresh token

otp_code async

otp_code(
    secret: str | bytes, factor: int | None = None
) -> str

Generates an OTP.

Parameters:

Name Type Description Default
secret str | bytes

User secret key.

required
factor int | None

Unixtime for TOTP(if none, use now time) / Counter for HOTP.

None

Returns:

Name Type Description
str str

OTP code (fixed-length string).

otp_uri async

otp_uri(
    secret: str,
    name: str,
    issuer: str,
    counter: int | None = None,
) -> str

Generates an otpauth:// URI for Google Authenticator.

Parameters:

Name Type Description Default
secret str

User secret key.

required
name str

Account name (e.g., email).

required
issuer str

Service name (e.g., "GitHub").

required
counter int | None

Counter (for HOTP). Default is None.

None

Returns:

Name Type Description
str str

A string of the form "otpauth://..."

otp_verify_code async

otp_verify_code(
    secret: str | bytes,
    code: str,
    factor: int | None = None,
    look_ahead: int | None = 1,
) -> bool

Checks the OTP code, taking into account the acceptable window.

Parameters:

Name Type Description Default
secret str | bytes

User secret key.

required
code str

The code entered.

required
factor int | None

Unixtime for TOTP(if none, use now time) / Counter for HOTP.

None
look_ahead int

Acceptable deviation in intervals (±window(totp) / ±look ahead(hotp)). Default is 1.

1

Returns:

Name Type Description
bool bool

True if the code matches, otherwise False.

paseto_create async

paseto_create(
    payload: dict[str, Any],
    footer: dict[str, Any] | str | None,
) -> str

Create new PASETO.

Parameters:

Name Type Description Default
payload dict[str, Any]

Payload

required
footer dict | str | None

Footer

required

Returns:

Name Type Description
str str

New token

paseto_decode async

paseto_decode(
    token: str,
    check_exp: bool = True,
    check_list: bool = True,
) -> dict[str, dict[str, Any] | str | None]

Decode PASETO.

Parameters:

Name Type Description Default
token str

Token

required
check_exp bool

Check exp in payload

True
check_list bool

Check token in list

True

Returns:

Name Type Description
dict dict[str, dict[str, Any] | str | None]

{'payload' PAYLOAD, 'footer': FOOTER}

paseto_make_payload async

paseto_make_payload(
    exp: int | None = None, **data: dict[str, Any]
) -> dict[str, Any]

Generate payload for PASETO.

Parameters:

Name Type Description Default
exp int | None

Token expire

None
data dict[str, Any]

Custom data

{}

Returns:

Name Type Description
dict dict[str, Any]

Payload

session_clear async

session_clear(session_key: str) -> None

Delete all sessions by key.

Parameters:

Name Type Description Default
session_key str

Key of session

required

session_create async

session_create(
    session_key: str, data: dict[str, Any]
) -> str

Create new session.

Parameters:

Name Type Description Default
session_key str

Key for session

required
data dict[str, Any]

Session data

required

Returns:

Name Type Description
str str

New session ID

session_delete async

session_delete(session_id: str) -> None

Delete session.

Parameters:

Name Type Description Default
session_id str

Session ID

required

session_get async

session_get(session_id: str) -> dict[str, Any] | None

Get data from session.

Parameters:

Name Type Description Default
session_id str

Session ID

required

Returns:

Type Description
dict[str, Any] | None

dict[str, Any] | None: Session data if exist

session_rework async

session_rework(old_session_id: str) -> str

Rework session.

Parameters:

Name Type Description Default
old_session_id str

Old session id

required

Returns:

Name Type Description
str str

New session id

session_update async

session_update(
    session_id: str, data: dict[str, Any]
) -> None

Update session data.

Parameters:

Name Type Description Default
session_id str

Session ID

required
data dict[str, Any]

New data

required