File size: 2,232 Bytes
b404de3 07a4a25 b404de3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | 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()
# # -------------------------------
# # Run server
# # -------------------------------
# def main():
# import uvicorn
# uvicorn.run("cloudenv.server.app:app")
# if __name__ == "__main__":
# main()
|