jam.aio.base¶
__base__
¶
Classes:
| Name | Description |
|---|---|
BaseAsyncJam |
Base async jam instance. |
BaseAsyncJam
¶
BaseAsyncJam(
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,
)
Bases: BaseJam
Base async jam instance.
Methods:
| Name | Description |
|---|---|
jwt_create |
Create JWT token. |
jwt_decode |
Verify and decode JWT token. |
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 and return payload and footer. |
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. |
jwt_create
abstractmethod
async
¶
jwt_create(payload: dict[str, Any]) -> str
Create JWT token.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payload
|
dict[str, Any]
|
Data payload |
required |
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
New token |
jwt_decode
abstractmethod
async
¶
jwt_decode(
token: str,
check_exp: bool = True,
check_list: bool = True,
) -> 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
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
dict[str, Any]: Decoded payload |
jwt_make_payload
abstractmethod
async
¶
jwt_make_payload(
exp: int | None, data: dict[str, Any]
) -> dict[str, Any]
Make JWT-specific payload.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exp
|
int | None
|
Token expire, if None -> use default |
required |
data
|
dict[str, Any]
|
Data to payload |
required |
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
dict[str, Any]: Payload |
oauth2_client_credentials_flow
abstractmethod
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
|
Provider name |
required |
scope
|
list[str] | None
|
Auth scope |
None
|
extra_params
|
Any
|
Extra auth params if needed |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
dict |
dict[str, Any]
|
JSON with access token |
oauth2_fetch_token
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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
abstractmethod
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, Any] | str | None
|
Payload if needed |
required |
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
PASETO |
paseto_decode
abstractmethod
async
¶
paseto_decode(
token: str,
check_exp: bool = True,
check_list: bool = True,
) -> dict[str, dict | str | dict | None]
Decode PASETO and return payload and footer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
PASETO |
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 | dict | None]
|
{ |
paseto_make_payload
abstractmethod
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
|
Custom expire if needed |
None
|
data
|
dict[str, Any]
|
Data in payload |
{}
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
dict[str, Any]: New payload |
session_clear
abstractmethod
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
abstractmethod
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
abstractmethod
async
¶
session_delete(session_id: str) -> None
Delete session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
session_id
|
str
|
Session ID |
required |
session_get
abstractmethod
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
abstractmethod
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
abstractmethod
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 |