| from fastapi import Request |
| from fastapi.responses import JSONResponse |
| from openenv.core.env_server.http_server import create_fastapi_app |
|
|
| from cloudenv.models import CloudEnvAction, CloudEnvObservation |
| from cloudenv.server.api.pipeline.api import router as PipelineApiRouter |
| from cloudenv.server.cloudenv_environment import CloudEnvironment |
| from cloudenv.server.exceptions import CloudEnvException |
| from cloudenv.server.managers.pipeline_manager import PipelineManager |
| from cloudenv.server.providers.aws.adapter import AWSAdapter |
| from cloudenv.server.providers.aws.audit_logger import AwsApiAuditLogger |
| from cloudenv.server.providers.aws.iam_enforcer import IAMEnforcer |
|
|
| audit_logger = AwsApiAuditLogger() |
| iam_enforcer = IAMEnforcer() |
|
|
| aws_adapter = AWSAdapter(audit_logger=audit_logger, iam_enforcer=iam_enforcer) |
| pipeline_manager = PipelineManager( |
| audit_logger=audit_logger, iam_enforcer=iam_enforcer |
| ) |
|
|
| cloud_env = CloudEnvironment( |
| aws=aws_adapter, |
| pipeline_manager=pipeline_manager, |
| ) |
|
|
|
|
| def cloud_environment_factory() -> CloudEnvironment: |
| return cloud_env |
|
|
|
|
| app = create_fastapi_app( |
| cloud_environment_factory, |
| CloudEnvAction, |
| CloudEnvObservation, |
| max_concurrent_envs=1, |
| ) |
|
|
| app.state.pipeline_manager = pipeline_manager |
| app.state.aws_adapter = aws_adapter |
| app.include_router(prefix="/api", router=PipelineApiRouter) |
|
|
|
|
| @app.exception_handler(CloudEnvException) |
| async def cloud_env_exception_handler(request: Request, exc: CloudEnvException): |
| return JSONResponse( |
| status_code=exc.status_code, |
| content={ |
| "error": exc.error_code, |
| "message": str(exc), |
| "detail": exc.detail, |
| }, |
| ) |
|
|
|
|
| def main(): |
| import argparse |
|
|
| import uvicorn |
|
|
| parser = argparse.ArgumentParser() |
| parser.add_argument("--host", type=str, default="0.0.0.0") |
| parser.add_argument("--port", type=int, default=8000) |
| args = parser.parse_args() |
|
|
| uvicorn.run(app, host=args.host, port=args.port) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
| |
| |
| |
| |
| |
| |
|
|
| |
| |
|
|
