| def switch_tenant(account: Account, tenant_id: int = None) -> None: | def switch_tenant(account: Account, tenant_id: int = None) -> None: | ||||
| """Switch the current workspace for the account""" | """Switch the current workspace for the account""" | ||||
| # Ensure tenant_id is provided | |||||
| if tenant_id is None: | |||||
| raise ValueError("Tenant ID must be provided.") | |||||
| tenant_account_join = TenantAccountJoin.query.filter_by(account_id=account.id, tenant_id=tenant_id).first() | tenant_account_join = TenantAccountJoin.query.filter_by(account_id=account.id, tenant_id=tenant_id).first() | ||||
| if not tenant_account_join: | if not tenant_account_join: | ||||
| raise AccountNotLinkTenantError("Tenant not found or account is not a member of the tenant.") | raise AccountNotLinkTenantError("Tenant not found or account is not a member of the tenant.") | ||||
| else: | else: | ||||
| with db.session.begin(): | |||||
| try: | |||||
| TenantAccountJoin.query.filter_by(account_id=account.id).update({'current': False}) | |||||
| tenant_account_join.current = True | |||||
| db.session.commit() | |||||
| # Set the current tenant for the account | |||||
| account.current_tenant_id = tenant_account_join.tenant_id | |||||
| except exc.SQLAlchemyError: | |||||
| db.session.rollback() | |||||
| raise | |||||
| TenantAccountJoin.query.filter(TenantAccountJoin.account_id == account.id, TenantAccountJoin.tenant_id != tenant_id).update({'current': False}) | |||||
| tenant_account_join.current = True | |||||
| db.session.commit() | |||||
| # Set the current tenant for the account | |||||
| account.current_tenant_id = tenant_account_join.tenant_id | |||||
| @staticmethod | @staticmethod | ||||
| def get_tenant_members(tenant: Tenant) -> List[Account]: | def get_tenant_members(tenant: Tenant) -> List[Account]: |