(app: FastAPI)
| 73 | |
| 74 | |
| 75 | def init_dynamic_cors(app: FastAPI): |
| 76 | try: |
| 77 | with Session(engine) as session: |
| 78 | list_result = session.exec(select(AssistantModel).order_by(AssistantModel.create_time)).all() |
| 79 | seen = set() |
| 80 | unique_domains = [] |
| 81 | for item in list_result: |
| 82 | if item.domain: |
| 83 | for domain in get_domain_list(item.domain): |
| 84 | domain = domain.strip() |
| 85 | if domain and domain not in seen: |
| 86 | seen.add(domain) |
| 87 | unique_domains.append(domain) |
| 88 | cors_middleware = None |
| 89 | response_middleware = None |
| 90 | for middleware in app.user_middleware: |
| 91 | if not cors_middleware and middleware.cls == CORSMiddleware: |
| 92 | cors_middleware = middleware |
| 93 | if not response_middleware and middleware.cls == ResponseMiddleware: |
| 94 | response_middleware = middleware |
| 95 | if cors_middleware and response_middleware: |
| 96 | break |
| 97 | |
| 98 | updated_origins = list(set(settings.all_cors_origins + unique_domains)) |
| 99 | if cors_middleware: |
| 100 | cors_middleware.kwargs['allow_origins'] = updated_origins |
| 101 | if response_middleware: |
| 102 | for instance in ResponseMiddleware.instances: |
| 103 | instance.update_allow_origins(updated_origins) |
| 104 | |
| 105 | except Exception as e: |
| 106 | return False, e |
| 107 | |
| 108 | |
| 109 | class AssistantOutDs: |
no test coverage detected