Skip to content

Commit

Permalink
more apis
Browse files Browse the repository at this point in the history
  • Loading branch information
rishabhpoddar committed Sep 16, 2024
1 parent 713efb5 commit da03488
Show file tree
Hide file tree
Showing 7 changed files with 557 additions and 0 deletions.
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
# Copyright (c) 2024, 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 typing import List, Dict, Any
from supertokens_python.recipe.multitenancy.recipe import MultitenancyRecipe
from supertokens_python.types import APIResponse
from ...interfaces import APIInterface, APIOptions
from .utils import (
get_normalised_first_factors_based_on_tenant_config_from_core_and_sdk_init,
)


class TenantWithLoginMethods:
def __init__(self, tenant_id: str, first_factors: List[str]):
self.tenant_id = tenant_id
self.first_factors = first_factors


class ListAllTenantsWithLoginMethodsOkResult(APIResponse):
def __init__(self, tenants: List[TenantWithLoginMethods]):
self.status = "OK"
self.tenants = tenants

def to_json(self) -> Dict[str, Any]:
return {
"status": self.status,
"tenants": [
{"tenantId": tenant.tenant_id, "firstFactors": tenant.first_factors}
for tenant in self.tenants
],
}


async def list_all_tenants_with_login_methods(
_: APIInterface,
__: str,
___: APIOptions,
user_context: Dict[str, Any],
) -> ListAllTenantsWithLoginMethodsOkResult:
tenants_res = (
await MultitenancyRecipe.get_instance().recipe_implementation.list_all_tenants(
user_context
)
)
final_tenants: List[TenantWithLoginMethods] = []

for current_tenant in tenants_res.tenants:
login_methods = (
get_normalised_first_factors_based_on_tenant_config_from_core_and_sdk_init(
current_tenant
)
)

final_tenants.append(
TenantWithLoginMethods(
tenant_id=current_tenant.tenant_id,
first_factors=login_methods,
)
)

return ListAllTenantsWithLoginMethodsOkResult(tenants=final_tenants)
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
# Copyright (c) 2024, 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 typing import Any, Dict, Union
from typing_extensions import Literal
from supertokens_python.exceptions import raise_bad_input_exception
from supertokens_python.recipe.multitenancy.interfaces import TenantConfigCreateOrUpdate
from supertokens_python.recipe.multitenancy.recipe import MultitenancyRecipe
from supertokens_python.types import APIResponse
from ...interfaces import APIInterface, APIOptions


class UpdateTenantCoreConfigOkResult(APIResponse):
status: Literal["OK"] = "OK"

def __init__(self):
self.status = "OK"

def to_json(self) -> Dict[str, Literal["OK"]]:
return {"status": self.status}


class UpdateTenantCoreConfigUnknownTenantErrorResult(APIResponse):
status: Literal["UNKNOWN_TENANT_ERROR"] = "UNKNOWN_TENANT_ERROR"

def __init__(self):
self.status = "UNKNOWN_TENANT_ERROR"

def to_json(self) -> Dict[str, Literal["UNKNOWN_TENANT_ERROR"]]:
return {"status": self.status}


class UpdateTenantCoreConfigInvalidConfigErrorResult(APIResponse):
status: Literal["INVALID_CONFIG_ERROR"] = "INVALID_CONFIG_ERROR"

def __init__(self, message: str):
self.status = "INVALID_CONFIG_ERROR"
self.message = message

def to_json(self) -> Dict[str, Union[Literal["INVALID_CONFIG_ERROR"], str]]:
return {"status": self.status, "message": self.message}


async def update_tenant_core_config(
_: APIInterface,
tenant_id: str,
options: APIOptions,
user_context: Dict[str, Any],
) -> Union[
UpdateTenantCoreConfigOkResult,
UpdateTenantCoreConfigUnknownTenantErrorResult,
UpdateTenantCoreConfigInvalidConfigErrorResult,
]:
request_body = await options.request.json()
if request_body is None:
raise_bad_input_exception("Request body is required")
name = request_body["name"]
value = request_body["value"]

mt_recipe = MultitenancyRecipe.get_instance()

tenant_res = await mt_recipe.recipe_implementation.get_tenant(
tenant_id=tenant_id, user_context=user_context
)
if tenant_res is None:
return UpdateTenantCoreConfigUnknownTenantErrorResult()

try:
await mt_recipe.recipe_implementation.create_or_update_tenant(
tenant_id=tenant_id,
config=TenantConfigCreateOrUpdate(
core_config={name: value},
),
user_context=user_context,
)
except Exception as err:
err_msg = str(err)
if (
"SuperTokens core threw an error for a " in err_msg
and "with status code: 400" in err_msg
):
return UpdateTenantCoreConfigInvalidConfigErrorResult(
message=err_msg.split(" and message: ")[1]
)
raise err

return UpdateTenantCoreConfigOkResult()
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
# Copyright (c) 2024, 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 typing import Any, Dict, Union
from typing_extensions import Literal
from supertokens_python.exceptions import raise_bad_input_exception
from supertokens_python.recipe.multitenancy.interfaces import TenantConfigCreateOrUpdate
from supertokens_python.recipe.multitenancy.recipe import MultitenancyRecipe
from supertokens_python.types import APIResponse
from ...interfaces import APIInterface, APIOptions
from .utils import (
get_factor_not_available_message,
get_normalised_first_factors_based_on_tenant_config_from_core_and_sdk_init,
)


class UpdateTenantFirstFactorOkResult(APIResponse):
status: Literal["OK"] = "OK"

def __init__(self):
self.status = "OK"

def to_json(self) -> Dict[str, Literal["OK"]]:
return {"status": self.status}


class UpdateTenantFirstFactorRecipeNotConfiguredOnBackendSdkErrorResult(APIResponse):
status: Literal[
"RECIPE_NOT_CONFIGURED_ON_BACKEND_SDK_ERROR"
] = "RECIPE_NOT_CONFIGURED_ON_BACKEND_SDK_ERROR"

def __init__(self, message: str):
self.status = "RECIPE_NOT_CONFIGURED_ON_BACKEND_SDK_ERROR"
self.message = message

def to_json(
self,
) -> Dict[str, Union[Literal["RECIPE_NOT_CONFIGURED_ON_BACKEND_SDK_ERROR"], str]]:
return {"status": self.status, "message": self.message}


class UpdateTenantFirstFactorUnknownTenantErrorResult(APIResponse):
status: Literal["UNKNOWN_TENANT_ERROR"] = "UNKNOWN_TENANT_ERROR"

def __init__(self):
self.status = "UNKNOWN_TENANT_ERROR"

def to_json(self) -> Dict[str, Literal["UNKNOWN_TENANT_ERROR"]]:
return {"status": self.status}


async def update_tenant_first_factor(
_: APIInterface,
tenant_id: str,
options: APIOptions,
user_context: Dict[str, Any],
) -> Union[
UpdateTenantFirstFactorOkResult,
UpdateTenantFirstFactorRecipeNotConfiguredOnBackendSdkErrorResult,
UpdateTenantFirstFactorUnknownTenantErrorResult,
]:
request_body = await options.request.json()
if request_body is None:
raise_bad_input_exception("Request body is required")
factor_id = request_body["factorId"]
enable = request_body["enable"]

mt_recipe = MultitenancyRecipe.get_instance()

if enable is True:
if factor_id not in mt_recipe.all_available_first_factors:
return UpdateTenantFirstFactorRecipeNotConfiguredOnBackendSdkErrorResult(
message=get_factor_not_available_message(
factor_id, mt_recipe.all_available_first_factors
)
)

tenant_res = await mt_recipe.recipe_implementation.get_tenant(
tenant_id=tenant_id, user_context=user_context
)

if tenant_res is None:
return UpdateTenantFirstFactorUnknownTenantErrorResult()

first_factors = (
get_normalised_first_factors_based_on_tenant_config_from_core_and_sdk_init(
tenant_res
)
)

if enable is True:
if factor_id not in first_factors:
first_factors.append(factor_id)
else:
first_factors = [f for f in first_factors if f != factor_id]

await mt_recipe.recipe_implementation.create_or_update_tenant(
tenant_id=tenant_id,
config=TenantConfigCreateOrUpdate(
first_factors=first_factors,
),
user_context=user_context,
)

return UpdateTenantFirstFactorOkResult()
Loading

0 comments on commit da03488

Please sign in to comment.