Merge pull request #1 from geobeyond/feature/pygeoapi

Feature/pygeoapi

.env

@@ -0,0 +1,47 @@
+ENV_STATE="prod" # or dev or prod
+
+# base configs
+# tiangolo uvicorn-gunicorn-fastapi-docker configs
+MODULE_NAME=app.main                            # or custom_app:custom_main
+VARIABLE_NAME=app                               # or some custom_var
+#- GUNICORN_CONF="/app/custom_gunicorn_conf.py"
+WORKERS_PER_CORE=1                              # by default 1
+WEB_CONCURRENCY=2                               # by default 2
+HOST=0.0.0.0                                    # by default 0.0.0.0
+PORT=5000                                       # by default 80
+LOG_LEVEL=info                                  # by default info
+#- WORKER_CLASS="uvicorn.workers.UvicornWorker"   # by default this. don't touch
+TIMEOUT=120                                     # by default 120 sec
+
+# aws
+AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
+AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
+AWS_DEPLOY=true
+AWS_LAMBDA_DEPLOY=true
+AWS_SM_ENDPOINT_URL="https://secretsmanager.eu-west-1.amazonaws.com"
+AWS_SM_SERVICE_NAME="secretsmanager"
+AWS_REGION_NAME="eu-west-1"
+
+# dev configs
+DEV_ROOT_PATH=
+DEV_AWS_LAMBDA_DEPLOY=false
+DEV_LOG_PATH="/tmp"
+DEV_LOG_FILENAME="fastgeoapi.log"
+DEV_LOG_LEVEL="debug"
+# loguru uses multiprocessing queue that breaks AWS lambda
+DEV_LOG_ENQUEUE=true
+DEV_LOG_ROTATION="1 days"
+DEV_LOG_RETENTION="1 months"
+DEV_LOG_FORMAT='<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> [id:{extra[request_id]}] - <level>{message}</level>'
+
+# prod configs
+PROD_ROOT_PATH=
+PROD_AWS_LAMBDA_DEPLOY=true
+PROD_LOG_PATH="/tmp"
+PROD_LOG_FILENAME="fastgeoapi.log"
+PROD_LOG_LEVEL="info"
+# loguru uses multiprocessing queue that breaks AWS lambda
+PROD_LOG_ENQUEUE=false
+PROD_LOG_ROTATION="1 days"
+PROD_LOG_RETENTION="1 months"
+PROD_LOG_FORMAT='<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> [id:{extra[request_id]}] - <level>{message}</level>'

.gitignore

@@ -6,5 +6,5 @@
 /.pytype/
 /dist/
 /docs_build/site/
-/src/*.egg-info/
+/app/*.egg-info/
 __pycache__/

.pre-commit-config.yaml

@@ -45,7 +45,7 @@ repos:
         entry: reorder-python-imports
         language: system
         types: [python]
-        args: [--application-directories=src]
+        args: [--application-directories=app]
       - id: trailing-whitespace
         name: Trim Trailing Whitespace
         entry: trailing-whitespace-fixer

app/__init__.py

@@ -0,0 +1 @@
+"""App module."""

app/config/__init__.py

@@ -0,0 +1 @@
+"""Config package."""

app/config/app.py

@@ -0,0 +1,66 @@
+"""App configuration module."""
+from typing import Optional
+
+from pydantic import BaseSettings
+from pydantic import Field
+
+
+class GlobalConfig(BaseSettings):
+    """Global configurations."""
+
+    # This variable will be loaded from the .env file. However, if there is a
+    # shell environment variable having the same name, that will take precedence.
+
+    ENV_STATE: Optional[str] = Field(None, env="ENV_STATE")
+
+    class Config:
+        """Loads the dotenv file."""
+
+        env_file: str = ".env"
+
+
+class DevConfig(GlobalConfig):
+    """Development configurations."""
+
+    ROOT_PATH: Optional[str] = Field(None, env="DEV_ROOT_PATH")
+    AWS_LAMBDA_DEPLOY: Optional[bool] = Field(None, env="DEV_AWS_LAMBDA_DEPLOY")
+    LOG_PATH: Optional[str] = Field(None, env="DEV_LOG_PATH")
+    LOG_FILENAME: Optional[str] = Field(None, env="DEV_LOG_FILENAME")
+    LOG_LEVEL: Optional[str] = Field(None, env="DEV_LOG_LEVEL")
+    LOG_ENQUEUE: Optional[bool] = Field(None, env="DEV_LOG_ENQUEUE")
+    LOG_ROTATION: Optional[str] = Field(None, env="DEV_LOG_ROTATION")
+    LOG_RETENTION: Optional[str] = Field(None, env="DEV_LOG_RETENTION")
+    LOG_FORMAT: Optional[str] = Field(None, env="DEV_LOG_FORMAT")
+
+
+class ProdConfig(GlobalConfig):
+    """Production configurations."""
+
+    ROOT_PATH: Optional[str] = Field(None, env="PROD_ROOT_PATH")
+    AWS_LAMBDA_DEPLOY: Optional[bool] = Field(None, env="PROD_AWS_LAMBDA_DEPLOY")
+    LOG_PATH: Optional[str] = Field(None, env="PROD_LOG_PATH")
+    LOG_FILENAME: Optional[str] = Field(None, env="PROD_LOG_FILENAME")
+    LOG_LEVEL: Optional[str] = Field(None, env="PROD_LOG_LEVEL")
+    LOG_ENQUEUE: Optional[bool] = Field(None, env="PROD_LOG_ENQUEUE")
+    LOG_ROTATION: Optional[str] = Field(None, env="PROD_LOG_ROTATION")
+    LOG_RETENTION: Optional[str] = Field(None, env="PROD_LOG_RETENTION")
+    LOG_FORMAT: Optional[str] = Field(None, env="PROD_LOG_FORMAT")
+
+
+class FactoryConfig:
+    """Returns a config instance dependending on the ENV_STATE variable."""
+
+    def __init__(self, env_state: Optional[str]):
+        """Initialize factory configuration."""
+        self.env_state = env_state
+
+    def __call__(self):
+        """Handle runtime configuration."""
+        if self.env_state == "dev":
+            return DevConfig()
+
+        elif self.env_state == "prod":
+            return ProdConfig()
+
+
+configuration = FactoryConfig(GlobalConfig().ENV_STATE)()

app/config/logging.py

@@ -0,0 +1,113 @@
+"""Logging module."""
+import logging
+import sys
+from pathlib import Path
+
+from app.config.app import configuration as cfg
+from app.schemas.logging import LoggerModel
+from app.schemas.logging import LoggingBase
+from loguru import logger
+
+
+class InterceptHandler(logging.Handler):
+    """Custom logging interceptor."""
+
+    loglevel_mapping = {
+        50: "CRITICAL",
+        40: "ERROR",
+        30: "WARNING",
+        20: "INFO",
+        10: "DEBUG",
+        0: "NOTSET",
+    }
+
+    def emit(self, record):
+        """Emits a logging record."""
+        try:
+            level = logger.level(record.levelname).name
+        except AttributeError:
+            level = self.loglevel_mapping[record.levelno]
+
+        frame, depth = logging.currentframe(), 2
+        while frame.f_code.co_filename == logging.__file__:
+            frame = frame.f_back
+            depth += 1
+
+        log = logger.bind(request_id="app")
+        log.opt(depth=depth, exception=record.exc_info).log(
+            level,
+            record.getMessage(),
+        )
+
+
+class CustomizeLogger:
+    """Handle logger customization."""
+
+    @classmethod
+    def make_logger(cls, config: LoggerModel):
+        """Create a logger instance."""
+        logging_config = config.logger
+
+        logger = cls.customize_logging(
+            filepath=logging_config.path,
+            level=logging_config.level,
+            enqueue=logging_config.enqueue,
+            retention=logging_config.retention,
+            rotation=logging_config.rotation,
+            format=logging_config.format_,
+        )
+        return logger
+
+    @classmethod
+    def customize_logging(
+        cls,
+        filepath: Path,
+        level: str,
+        enqueue: bool,
+        rotation: str,
+        retention: str,
+        format: str,
+    ):
+        """Customize logging configuration."""
+        logger.remove()
+        logger.add(
+            sys.stdout,
+            enqueue=enqueue,
+            backtrace=True,
+            level=level.upper(),
+            format=format,
+        )
+        logger.add(
+            str(filepath),
+            rotation=rotation,
+            retention=retention,
+            enqueue=enqueue,
+            backtrace=True,
+            level=level.upper(),
+            format=format,
+        )
+        logging.basicConfig(handlers=[InterceptHandler()], level=0)
+        logging.getLogger("uvicorn.access").handlers = [InterceptHandler()]
+        for _log in ["uvicorn", "uvicorn.error", "fastapi"]:
+            _logger = logging.getLogger(_log)
+            _logger.handlers = [InterceptHandler()]
+
+        return logger.bind(request_id=None, method=None)
+
+
+def create_logger(name: str):
+    """Create a logger instance."""
+    logger = logging.getLogger(name)
+    config = LoggerModel(
+        logger=LoggingBase(
+            path=Path(cfg.LOG_PATH) / cfg.LOG_FILENAME,
+            level=cfg.LOG_LEVEL,
+            enqueue=cfg.LOG_ENQUEUE,
+            retention=cfg.LOG_RETENTION,
+            rotation=cfg.LOG_ROTATION,
+            format_=cfg.LOG_FORMAT,
+        ),
+    )
+    logger = CustomizeLogger.make_logger(config)
+
+    return logger

app/main.py

@@ -0,0 +1,60 @@
+"""Main module."""
+import uvicorn
+from app.config.app import configuration as cfg
+from app.config.logging import create_logger
+from app.utils.app_exceptions import app_exception_handler
+from app.utils.app_exceptions import AppExceptionError
+from app.utils.request_exceptions import http_exception_handler
+from app.utils.request_exceptions import request_validation_exception_handler
+from fastapi import FastAPI
+from fastapi.exceptions import RequestValidationError
+from mangum import Mangum
+from starlette.exceptions import HTTPException as StarletteHTTPException
+from starlette.middleware.cors import CORSMiddleware
+
+from pygeoapi.starlette_app import app as pygeoapi_app
+
+
+def create_app() -> FastAPI:
+    """Handle application creation."""
+    app = FastAPI(title="Fastgeoapi", root_path=cfg.ROOT_PATH, debug=True)
+
+    # Set all CORS enabled origins
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.exception_handler(StarletteHTTPException)
+    async def custom_http_exception_handler(request, e):
+        return await http_exception_handler(request, e)
+
+    @app.exception_handler(RequestValidationError)
+    async def custom_validation_exception_handler(request, e):
+        return await request_validation_exception_handler(request, e)
+
+    @app.exception_handler(AppExceptionError)
+    async def custom_app_exception_handler(request, e):
+        return await app_exception_handler(request, e)
+
+    app.mount(path="/api", app=pygeoapi_app)
+
+    app.logger = create_logger(name="app.main")
+
+    return app
+
+
+app = create_app()
+
+app.logger.debug(f"Global config: {cfg.__repr__()}")
+
+if cfg.AWS_LAMBDA_DEPLOY:
+    # to make it work with Amazon Lambda,
+    # we create a handler object
+    handler = Mangum(app)
+
+if __name__ == "__main__":
+    uvicorn.run(app, port=5000)

app/schemas/logging.py

@@ -0,0 +1,21 @@
+"""Logging module."""
+from pathlib import Path
+
+from pydantic import BaseModel
+
+
+class LoggingBase(BaseModel):
+    """Base logging model."""
+
+    path: Path
+    level: str
+    enqueue: bool
+    retention: str
+    rotation: str
+    format_: str
+
+
+class LoggerModel(BaseModel):
+    """Logger model."""
+
+    logger: LoggingBase

app/utils/__init__.py

@@ -0,0 +1 @@
+"""Utils module."""

app/utils/app_exceptions.py

@@ -0,0 +1,38 @@
+"""App exceptions module."""
+from fastapi import Request
+from starlette.responses import JSONResponse
+
+
+class AppExceptionError(Exception):
+    """Application exception base error class."""
+
+    def __init__(self, status_code: int, error: str, context: dict):
+        """Handle application exceptions initialization."""
+        self.exception_case = self.__class__.__name__
+        self.status_code = status_code
+        self.error = error
+        self.context = context
+
+    def __str__(self):
+        """Define representation of application exception instance."""
+        return (
+            f"<AppExceptionError {self.exception_case} - error={self.error}"
+            + f"status_code={self.status_code} - context={self.context}>"
+        )
+
+
+async def app_exception_handler(request: Request, exc: AppExceptionError):
+    """Handle json representation of application exception."""
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={
+            "error": exc.error,
+            "context": exc.context,
+        },
+    )
+
+
+class AppException:
+    """Application exception class."""
+
+    pass

app/utils/get_list_of_app_exceptions_for_frontend.py

@@ -0,0 +1,11 @@
+"""App exceptions for frontend module."""
+from app.config.logging import create_logger
+from app.utils.app_exceptions import AppException
+
+
+logger = create_logger(
+    name="app.utils.get_list_of_app_exceptions_for_frontend",
+)
+
+
+logger.debug([e for e in dir(AppException) if "__" not in e])