Module supertokens_python.recipe.session.syncio
Expand source code
# Copyright (c) 2021, VRAI Labs and/or its affiliates. All rights reserved.
#
# This software is licensed under the Apache License, Version 2.0 (the
# "License") as published by the Apache Software Foundation.
#
# You may not use this file except in compliance with the License. You may
# obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import annotations
from typing import Any, Dict, List, Union, Callable, Optional, TypeVar
from supertokens_python.async_to_sync_wrapper import sync
from supertokens_python.recipe.openid.interfaces import (
GetOpenIdDiscoveryConfigurationResult,
)
from supertokens_python.types import MaybeAwaitable
from ...jwt.interfaces import (
CreateJwtOkResult,
CreateJwtResultUnsupportedAlgorithm,
GetJWKSResult,
)
from ..interfaces import (
SessionContainer,
SessionInformationResult,
SessionClaimValidator,
SessionClaim,
ClaimsValidationResult,
SessionDoesNotExistError,
GetClaimValueOkResult,
)
from supertokens_python.recipe.session.recipe_implementation import RecipeUserId
def create_new_session(
request: Any,
tenant_id: str,
recipe_user_id: RecipeUserId,
access_token_payload: Union[Dict[str, Any], None] = None,
session_data_in_database: Union[Dict[str, Any], None] = None,
user_context: Union[None, Dict[str, Any]] = None,
) -> SessionContainer:
from supertokens_python.recipe.session.asyncio import (
create_new_session as async_create_new_session,
)
return sync(
async_create_new_session(
request=request,
tenant_id=tenant_id,
recipe_user_id=recipe_user_id,
access_token_payload=access_token_payload,
session_data_in_database=session_data_in_database,
user_context=user_context,
)
)
def create_new_session_without_request_response(
tenant_id: str,
recipe_user_id: RecipeUserId,
access_token_payload: Union[Dict[str, Any], None] = None,
session_data_in_database: Union[Dict[str, Any], None] = None,
disable_anti_csrf: bool = False,
user_context: Union[None, Dict[str, Any]] = None,
) -> SessionContainer:
from supertokens_python.recipe.session.asyncio import (
create_new_session_without_request_response as async_create_new_session_without_request_response,
)
return sync(
async_create_new_session_without_request_response(
tenant_id,
recipe_user_id,
access_token_payload,
session_data_in_database,
disable_anti_csrf,
user_context,
)
)
def get_session(
request: Any,
session_required: bool = True,
anti_csrf_check: Union[bool, None] = None,
check_database: Optional[bool] = None,
override_global_claim_validators: Optional[
Callable[
[List[SessionClaimValidator], SessionContainer, Dict[str, Any]],
MaybeAwaitable[List[SessionClaimValidator]],
]
] = None,
user_context: Union[None, Dict[str, Any]] = None,
) -> Union[SessionContainer, None]:
from supertokens_python.recipe.session.asyncio import (
get_session as async_get_session,
)
return sync(
async_get_session(
request,
session_required,
anti_csrf_check,
check_database,
override_global_claim_validators,
user_context,
)
)
def get_session_without_request_response(
access_token: str,
anti_csrf_token: Optional[str] = None,
anti_csrf_check: Optional[bool] = None,
session_required: Optional[bool] = None,
check_database: Optional[bool] = None,
override_global_claim_validators: Optional[
Callable[
[List[SessionClaimValidator], SessionContainer, Dict[str, Any]],
MaybeAwaitable[List[SessionClaimValidator]],
]
] = None,
user_context: Union[None, Dict[str, Any]] = None,
) -> Optional[SessionContainer]:
from supertokens_python.recipe.session.asyncio import (
get_session_without_request_response as async_get_session_without_request_response,
)
return sync(
async_get_session_without_request_response(
access_token,
anti_csrf_token,
anti_csrf_check,
session_required,
check_database,
override_global_claim_validators,
user_context,
)
)
def refresh_session(
request: Any, user_context: Union[None, Dict[str, Any]] = None
) -> SessionContainer:
from supertokens_python.recipe.session.asyncio import (
refresh_session as async_refresh_session,
)
return sync(async_refresh_session(request, user_context))
def refresh_session_without_request_response(
refresh_token: str,
disable_anti_csrf: bool = False,
anti_csrf_token: Optional[str] = None,
user_context: Optional[Dict[str, Any]] = None,
) -> SessionContainer:
from supertokens_python.recipe.session.asyncio import (
refresh_session_without_request_response as async_refresh_session_without_request_response,
)
return sync(
async_refresh_session_without_request_response(
refresh_token,
disable_anti_csrf,
anti_csrf_token,
user_context,
)
)
def revoke_session(
session_handle: str, user_context: Union[None, Dict[str, Any]] = None
) -> bool:
from supertokens_python.recipe.session.asyncio import (
revoke_session as async_revoke_session,
)
return sync(async_revoke_session(session_handle, user_context))
def revoke_all_sessions_for_user(
user_id: str,
revoke_sessions_for_linked_accounts: bool = True,
tenant_id: Optional[str] = None,
user_context: Union[None, Dict[str, Any]] = None,
) -> List[str]:
from supertokens_python.recipe.session.asyncio import (
revoke_all_sessions_for_user as async_revoke_all_sessions_for_user,
)
return sync(
async_revoke_all_sessions_for_user(
user_id,
revoke_sessions_for_linked_accounts,
tenant_id,
user_context,
)
)
def get_all_session_handles_for_user(
user_id: str,
fetch_sessions_for_linked_accounts: bool = True,
tenant_id: Optional[str] = None,
user_context: Union[None, Dict[str, Any]] = None,
) -> List[str]:
from supertokens_python.recipe.session.asyncio import (
get_all_session_handles_for_user as async_get_all_session_handles_for_user,
)
return sync(
async_get_all_session_handles_for_user(
user_id,
fetch_sessions_for_linked_accounts,
tenant_id,
user_context,
)
)
def revoke_multiple_sessions(
session_handles: List[str], user_context: Union[None, Dict[str, Any]] = None
) -> List[str]:
from supertokens_python.recipe.session.asyncio import (
revoke_multiple_sessions as async_revoke_multiple_sessions,
)
return sync(async_revoke_multiple_sessions(session_handles, user_context))
def get_session_information(
session_handle: str, user_context: Union[None, Dict[str, Any]] = None
) -> Union[SessionInformationResult, None]:
from supertokens_python.recipe.session.asyncio import (
get_session_information as async_get_session_information,
)
return sync(async_get_session_information(session_handle, user_context))
def update_session_data_in_database(
session_handle: str,
new_session_data: Dict[str, Any],
user_context: Union[None, Dict[str, Any]] = None,
) -> bool:
from supertokens_python.recipe.session.asyncio import (
update_session_data_in_database as async_update_session_data_in_database,
)
return sync(
async_update_session_data_in_database(
session_handle, new_session_data, user_context
)
)
def merge_into_access_token_payload(
session_handle: str,
new_access_token_payload: Dict[str, Any],
user_context: Union[None, Dict[str, Any]] = None,
) -> bool:
from supertokens_python.recipe.session.asyncio import (
merge_into_access_token_payload as async_merge_into_access_token_payload,
)
return sync(
async_merge_into_access_token_payload(
session_handle, new_access_token_payload, user_context
)
)
def create_jwt(
payload: Dict[str, Any],
validity_seconds: Optional[int] = None,
use_static_signing_key: Optional[bool] = None,
user_context: Optional[Dict[str, Any]] = None,
) -> Union[CreateJwtOkResult, CreateJwtResultUnsupportedAlgorithm]:
from supertokens_python.recipe.session.asyncio import create_jwt as async_create_jwt
return sync(
async_create_jwt(
payload, validity_seconds, use_static_signing_key, user_context
)
)
def get_jwks(user_context: Optional[Dict[str, Any]] = None) -> GetJWKSResult:
from supertokens_python.recipe.session.asyncio import get_jwks as async_get_jwks
return sync(async_get_jwks(user_context))
def get_open_id_discovery_configuration(
user_context: Optional[Dict[str, Any]] = None
) -> GetOpenIdDiscoveryConfigurationResult:
from supertokens_python.recipe.session.asyncio import (
get_open_id_discovery_configuration as async_get_open_id_discovery_configuration,
)
return sync(async_get_open_id_discovery_configuration(user_context))
_T = TypeVar("_T")
def fetch_and_set_claim(
session_handle: str,
claim: SessionClaim[Any],
user_context: Union[None, Dict[str, Any]] = None,
) -> bool:
from supertokens_python.recipe.session.asyncio import (
fetch_and_set_claim as async_fetch_and_set_claim,
)
return sync(async_fetch_and_set_claim(session_handle, claim, user_context))
def set_claim_value(
session_handle: str,
claim: SessionClaim[_T],
value: _T,
user_context: Union[None, Dict[str, Any]] = None,
) -> bool:
from supertokens_python.recipe.session.asyncio import (
set_claim_value as async_set_claim_value,
)
return sync(async_set_claim_value(session_handle, claim, value, user_context))
def get_claim_value(
session_handle: str,
claim: SessionClaim[_T],
user_context: Union[None, Dict[str, Any]] = None,
) -> Union[SessionDoesNotExistError, GetClaimValueOkResult[_T]]:
from supertokens_python.recipe.session.asyncio import (
get_claim_value as async_get_claim_value,
)
return sync(async_get_claim_value(session_handle, claim, user_context))
def remove_claim(
session_handle: str,
claim: SessionClaim[Any],
user_context: Union[None, Dict[str, Any]] = None,
) -> bool:
from supertokens_python.recipe.session.asyncio import (
remove_claim as async_remove_claim,
)
return sync(async_remove_claim(session_handle, claim, user_context))
def validate_claims_for_session_handle(
session_handle: str,
override_global_claim_validators: Optional[
Callable[
[List[SessionClaimValidator], SessionInformationResult, Dict[str, Any]],
MaybeAwaitable[List[SessionClaimValidator]],
]
] = None,
user_context: Union[None, Dict[str, Any]] = None,
) -> Union[SessionDoesNotExistError, ClaimsValidationResult]:
from supertokens_python.recipe.session.asyncio import (
validate_claims_for_session_handle as async_validate_claims_for_session_handle,
)
return sync(
async_validate_claims_for_session_handle(
session_handle, override_global_claim_validators, user_context
)
)
Functions
def create_jwt(payload: Dict[str, Any], validity_seconds: Optional[int] = None, use_static_signing_key: Optional[bool] = None, user_context: Optional[Dict[str, Any]] = None) ‑> Union[CreateJwtOkResult, CreateJwtResultUnsupportedAlgorithm]
-
Expand source code
def create_jwt( payload: Dict[str, Any], validity_seconds: Optional[int] = None, use_static_signing_key: Optional[bool] = None, user_context: Optional[Dict[str, Any]] = None, ) -> Union[CreateJwtOkResult, CreateJwtResultUnsupportedAlgorithm]: from supertokens_python.recipe.session.asyncio import create_jwt as async_create_jwt return sync( async_create_jwt( payload, validity_seconds, use_static_signing_key, user_context ) )
def create_new_session(request: Any, tenant_id: str, recipe_user_id: RecipeUserId, access_token_payload: Union[Dict[str, Any], None] = None, session_data_in_database: Union[Dict[str, Any], None] = None, user_context: Union[None, Dict[str, Any]] = None) ‑> SessionContainer
-
Expand source code
def create_new_session( request: Any, tenant_id: str, recipe_user_id: RecipeUserId, access_token_payload: Union[Dict[str, Any], None] = None, session_data_in_database: Union[Dict[str, Any], None] = None, user_context: Union[None, Dict[str, Any]] = None, ) -> SessionContainer: from supertokens_python.recipe.session.asyncio import ( create_new_session as async_create_new_session, ) return sync( async_create_new_session( request=request, tenant_id=tenant_id, recipe_user_id=recipe_user_id, access_token_payload=access_token_payload, session_data_in_database=session_data_in_database, user_context=user_context, ) )
def create_new_session_without_request_response(tenant_id: str, recipe_user_id: RecipeUserId, access_token_payload: Union[Dict[str, Any], None] = None, session_data_in_database: Union[Dict[str, Any], None] = None, disable_anti_csrf: bool = False, user_context: Union[None, Dict[str, Any]] = None) ‑> SessionContainer
-
Expand source code
def create_new_session_without_request_response( tenant_id: str, recipe_user_id: RecipeUserId, access_token_payload: Union[Dict[str, Any], None] = None, session_data_in_database: Union[Dict[str, Any], None] = None, disable_anti_csrf: bool = False, user_context: Union[None, Dict[str, Any]] = None, ) -> SessionContainer: from supertokens_python.recipe.session.asyncio import ( create_new_session_without_request_response as async_create_new_session_without_request_response, ) return sync( async_create_new_session_without_request_response( tenant_id, recipe_user_id, access_token_payload, session_data_in_database, disable_anti_csrf, user_context, ) )
def fetch_and_set_claim(session_handle: str, claim: SessionClaim[Any], user_context: Union[None, Dict[str, Any]] = None) ‑> bool
-
Expand source code
def fetch_and_set_claim( session_handle: str, claim: SessionClaim[Any], user_context: Union[None, Dict[str, Any]] = None, ) -> bool: from supertokens_python.recipe.session.asyncio import ( fetch_and_set_claim as async_fetch_and_set_claim, ) return sync(async_fetch_and_set_claim(session_handle, claim, user_context))
def get_all_session_handles_for_user(user_id: str, fetch_sessions_for_linked_accounts: bool = True, tenant_id: Optional[str] = None, user_context: Union[None, Dict[str, Any]] = None) ‑> List[str]
-
Expand source code
def get_all_session_handles_for_user( user_id: str, fetch_sessions_for_linked_accounts: bool = True, tenant_id: Optional[str] = None, user_context: Union[None, Dict[str, Any]] = None, ) -> List[str]: from supertokens_python.recipe.session.asyncio import ( get_all_session_handles_for_user as async_get_all_session_handles_for_user, ) return sync( async_get_all_session_handles_for_user( user_id, fetch_sessions_for_linked_accounts, tenant_id, user_context, ) )
def get_claim_value(session_handle: str, claim: SessionClaim[_T], user_context: Union[None, Dict[str, Any]] = None) ‑> Union[SessionDoesNotExistError, GetClaimValueOkResult[~_T]]
-
Expand source code
def get_claim_value( session_handle: str, claim: SessionClaim[_T], user_context: Union[None, Dict[str, Any]] = None, ) -> Union[SessionDoesNotExistError, GetClaimValueOkResult[_T]]: from supertokens_python.recipe.session.asyncio import ( get_claim_value as async_get_claim_value, ) return sync(async_get_claim_value(session_handle, claim, user_context))
def get_jwks(user_context: Optional[Dict[str, Any]] = None) ‑> GetJWKSResult
-
Expand source code
def get_jwks(user_context: Optional[Dict[str, Any]] = None) -> GetJWKSResult: from supertokens_python.recipe.session.asyncio import get_jwks as async_get_jwks return sync(async_get_jwks(user_context))
def get_open_id_discovery_configuration(user_context: Optional[Dict[str, Any]] = None) ‑> GetOpenIdDiscoveryConfigurationResult
-
Expand source code
def get_open_id_discovery_configuration( user_context: Optional[Dict[str, Any]] = None ) -> GetOpenIdDiscoveryConfigurationResult: from supertokens_python.recipe.session.asyncio import ( get_open_id_discovery_configuration as async_get_open_id_discovery_configuration, ) return sync(async_get_open_id_discovery_configuration(user_context))
def get_session(request: Any, session_required: bool = True, anti_csrf_check: Union[bool, None] = None, check_database: Optional[bool] = None, override_global_claim_validators: Optional[Callable[[List[SessionClaimValidator], SessionContainer, Dict[str, Any]], MaybeAwaitable[List[SessionClaimValidator]]]] = None, user_context: Union[None, Dict[str, Any]] = None) ‑> Optional[SessionContainer]
-
Expand source code
def get_session( request: Any, session_required: bool = True, anti_csrf_check: Union[bool, None] = None, check_database: Optional[bool] = None, override_global_claim_validators: Optional[ Callable[ [List[SessionClaimValidator], SessionContainer, Dict[str, Any]], MaybeAwaitable[List[SessionClaimValidator]], ] ] = None, user_context: Union[None, Dict[str, Any]] = None, ) -> Union[SessionContainer, None]: from supertokens_python.recipe.session.asyncio import ( get_session as async_get_session, ) return sync( async_get_session( request, session_required, anti_csrf_check, check_database, override_global_claim_validators, user_context, ) )
def get_session_information(session_handle: str, user_context: Union[None, Dict[str, Any]] = None) ‑> Optional[SessionInformationResult]
-
Expand source code
def get_session_information( session_handle: str, user_context: Union[None, Dict[str, Any]] = None ) -> Union[SessionInformationResult, None]: from supertokens_python.recipe.session.asyncio import ( get_session_information as async_get_session_information, ) return sync(async_get_session_information(session_handle, user_context))
def get_session_without_request_response(access_token: str, anti_csrf_token: Optional[str] = None, anti_csrf_check: Optional[bool] = None, session_required: Optional[bool] = None, check_database: Optional[bool] = None, override_global_claim_validators: Optional[Callable[[List[SessionClaimValidator], SessionContainer, Dict[str, Any]], MaybeAwaitable[List[SessionClaimValidator]]]] = None, user_context: Union[None, Dict[str, Any]] = None) ‑> Optional[SessionContainer]
-
Expand source code
def get_session_without_request_response( access_token: str, anti_csrf_token: Optional[str] = None, anti_csrf_check: Optional[bool] = None, session_required: Optional[bool] = None, check_database: Optional[bool] = None, override_global_claim_validators: Optional[ Callable[ [List[SessionClaimValidator], SessionContainer, Dict[str, Any]], MaybeAwaitable[List[SessionClaimValidator]], ] ] = None, user_context: Union[None, Dict[str, Any]] = None, ) -> Optional[SessionContainer]: from supertokens_python.recipe.session.asyncio import ( get_session_without_request_response as async_get_session_without_request_response, ) return sync( async_get_session_without_request_response( access_token, anti_csrf_token, anti_csrf_check, session_required, check_database, override_global_claim_validators, user_context, ) )
def merge_into_access_token_payload(session_handle: str, new_access_token_payload: Dict[str, Any], user_context: Union[None, Dict[str, Any]] = None) ‑> bool
-
Expand source code
def merge_into_access_token_payload( session_handle: str, new_access_token_payload: Dict[str, Any], user_context: Union[None, Dict[str, Any]] = None, ) -> bool: from supertokens_python.recipe.session.asyncio import ( merge_into_access_token_payload as async_merge_into_access_token_payload, ) return sync( async_merge_into_access_token_payload( session_handle, new_access_token_payload, user_context ) )
def refresh_session(request: Any, user_context: Union[None, Dict[str, Any]] = None) ‑> SessionContainer
-
Expand source code
def refresh_session( request: Any, user_context: Union[None, Dict[str, Any]] = None ) -> SessionContainer: from supertokens_python.recipe.session.asyncio import ( refresh_session as async_refresh_session, ) return sync(async_refresh_session(request, user_context))
def refresh_session_without_request_response(refresh_token: str, disable_anti_csrf: bool = False, anti_csrf_token: Optional[str] = None, user_context: Optional[Dict[str, Any]] = None) ‑> SessionContainer
-
Expand source code
def refresh_session_without_request_response( refresh_token: str, disable_anti_csrf: bool = False, anti_csrf_token: Optional[str] = None, user_context: Optional[Dict[str, Any]] = None, ) -> SessionContainer: from supertokens_python.recipe.session.asyncio import ( refresh_session_without_request_response as async_refresh_session_without_request_response, ) return sync( async_refresh_session_without_request_response( refresh_token, disable_anti_csrf, anti_csrf_token, user_context, ) )
def remove_claim(session_handle: str, claim: SessionClaim[Any], user_context: Union[None, Dict[str, Any]] = None) ‑> bool
-
Expand source code
def remove_claim( session_handle: str, claim: SessionClaim[Any], user_context: Union[None, Dict[str, Any]] = None, ) -> bool: from supertokens_python.recipe.session.asyncio import ( remove_claim as async_remove_claim, ) return sync(async_remove_claim(session_handle, claim, user_context))
def revoke_all_sessions_for_user(user_id: str, revoke_sessions_for_linked_accounts: bool = True, tenant_id: Optional[str] = None, user_context: Union[None, Dict[str, Any]] = None) ‑> List[str]
-
Expand source code
def revoke_all_sessions_for_user( user_id: str, revoke_sessions_for_linked_accounts: bool = True, tenant_id: Optional[str] = None, user_context: Union[None, Dict[str, Any]] = None, ) -> List[str]: from supertokens_python.recipe.session.asyncio import ( revoke_all_sessions_for_user as async_revoke_all_sessions_for_user, ) return sync( async_revoke_all_sessions_for_user( user_id, revoke_sessions_for_linked_accounts, tenant_id, user_context, ) )
def revoke_multiple_sessions(session_handles: List[str], user_context: Union[None, Dict[str, Any]] = None) ‑> List[str]
-
Expand source code
def revoke_multiple_sessions( session_handles: List[str], user_context: Union[None, Dict[str, Any]] = None ) -> List[str]: from supertokens_python.recipe.session.asyncio import ( revoke_multiple_sessions as async_revoke_multiple_sessions, ) return sync(async_revoke_multiple_sessions(session_handles, user_context))
def revoke_session(session_handle: str, user_context: Union[None, Dict[str, Any]] = None) ‑> bool
-
Expand source code
def revoke_session( session_handle: str, user_context: Union[None, Dict[str, Any]] = None ) -> bool: from supertokens_python.recipe.session.asyncio import ( revoke_session as async_revoke_session, ) return sync(async_revoke_session(session_handle, user_context))
def set_claim_value(session_handle: str, claim: SessionClaim[_T], value: _T, user_context: Union[None, Dict[str, Any]] = None) ‑> bool
-
Expand source code
def set_claim_value( session_handle: str, claim: SessionClaim[_T], value: _T, user_context: Union[None, Dict[str, Any]] = None, ) -> bool: from supertokens_python.recipe.session.asyncio import ( set_claim_value as async_set_claim_value, ) return sync(async_set_claim_value(session_handle, claim, value, user_context))
def update_session_data_in_database(session_handle: str, new_session_data: Dict[str, Any], user_context: Union[None, Dict[str, Any]] = None) ‑> bool
-
Expand source code
def update_session_data_in_database( session_handle: str, new_session_data: Dict[str, Any], user_context: Union[None, Dict[str, Any]] = None, ) -> bool: from supertokens_python.recipe.session.asyncio import ( update_session_data_in_database as async_update_session_data_in_database, ) return sync( async_update_session_data_in_database( session_handle, new_session_data, user_context ) )
def validate_claims_for_session_handle(session_handle: str, override_global_claim_validators: Optional[Callable[[List[SessionClaimValidator], SessionInformationResult, Dict[str, Any]], MaybeAwaitable[List[SessionClaimValidator]]]] = None, user_context: Union[None, Dict[str, Any]] = None) ‑> Union[SessionDoesNotExistError, ClaimsValidationResult]
-
Expand source code
def validate_claims_for_session_handle( session_handle: str, override_global_claim_validators: Optional[ Callable[ [List[SessionClaimValidator], SessionInformationResult, Dict[str, Any]], MaybeAwaitable[List[SessionClaimValidator]], ] ] = None, user_context: Union[None, Dict[str, Any]] = None, ) -> Union[SessionDoesNotExistError, ClaimsValidationResult]: from supertokens_python.recipe.session.asyncio import ( validate_claims_for_session_handle as async_validate_claims_for_session_handle, ) return sync( async_validate_claims_for_session_handle( session_handle, override_global_claim_validators, user_context ) )