mirror of
https://github.com/langgenius/dify.git
synced 2026-05-28 21:03:22 +08:00
68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from werkzeug.exceptions import Forbidden, InternalServerError, NotFound, Unauthorized
|
|
|
|
from controllers.openapi.auth.data import AuthData
|
|
from core.app.entities.app_invoke_entities import InvokeFrom
|
|
from extensions.ext_database import db
|
|
from models.account import TenantStatus
|
|
from services.account_service import AccountService, TenantService
|
|
from services.app_service import AppService
|
|
from services.end_user_service import EndUserService
|
|
from services.enterprise.enterprise_service import EnterpriseService, WebAppAccessMode
|
|
|
|
|
|
def load_app(data: AuthData) -> None:
|
|
app_id = data.path_params["app_id"]
|
|
app = AppService.get_app_by_id(db.session, app_id)
|
|
if not app or app.status != "normal":
|
|
raise NotFound("app not found")
|
|
if not app.enable_api:
|
|
raise Forbidden("service_api_disabled")
|
|
data.app = app
|
|
|
|
|
|
def load_tenant(data: AuthData) -> None:
|
|
if data.app is None:
|
|
raise InternalServerError("pipeline_invariant_violated: app not loaded before load_tenant")
|
|
tenant = TenantService.get_tenant_by_id(db.session, str(data.app.tenant_id))
|
|
if tenant is None or tenant.status == TenantStatus.ARCHIVE:
|
|
raise Forbidden("workspace unavailable")
|
|
data.tenant = tenant
|
|
|
|
|
|
def load_account(data: AuthData) -> None:
|
|
account = AccountService.get_account_by_id(db.session, str(data.account_id))
|
|
if account is None:
|
|
raise Unauthorized("account not found")
|
|
if data.tenant:
|
|
account.current_tenant = data.tenant
|
|
data.caller = account
|
|
data.caller_kind = "account"
|
|
|
|
|
|
def resolve_external_user(data: AuthData) -> None:
|
|
if data.tenant is None or data.app is None or data.external_identity is None:
|
|
raise Unauthorized("missing context for external user resolution")
|
|
end_user = EndUserService.get_or_create_end_user_by_type(
|
|
InvokeFrom.OPENAPI,
|
|
tenant_id=str(data.tenant.id),
|
|
app_id=str(data.app.id),
|
|
user_id=data.external_identity.email,
|
|
)
|
|
data.caller = end_user
|
|
data.caller_kind = "end_user"
|
|
|
|
|
|
def load_app_access_mode(data: AuthData) -> None:
|
|
if data.app is None:
|
|
return
|
|
try:
|
|
settings = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id=str(data.app.id))
|
|
if settings is None:
|
|
data.app_access_mode = None
|
|
return
|
|
data.app_access_mode = WebAppAccessMode(settings.access_mode)
|
|
except ValueError:
|
|
data.app_access_mode = None
|