Source code for terracotta.config


Terracotta settings parsing.

from typing import Mapping, Any, Tuple, NamedTuple, Dict, List, Optional
import os
import json
import tempfile

from marshmallow import Schema, fields, validate, pre_load, post_load, ValidationError

[docs]class TerracottaSettings(NamedTuple): """Contains all settings for the current Terracotta instance.""" #: Path to database DRIVER_PATH: str = '' #: Driver provider to use (sqlite, sqlite-remote, mysql; auto-detected by default) DRIVER_PROVIDER: Optional[str] = None #: Activate debug mode in Flask app DEBUG: bool = False #: Print profile information after every request FLASK_PROFILE: bool = False #: Send performance traces to AWS X-Ray XRAY_PROFILE: bool = False #: Default log level (debug, info, warning, error, critical) LOGLEVEL: str = 'warning' #: Size of raster file in-memory cache in bytes RASTER_CACHE_SIZE: int = 1024 * 1024 * 490 # 490 MB #: Compression level of raster file in-memory cache, from 0-9 RASTER_CACHE_COMPRESS_LEVEL: int = 9 #: Tile size to return if not given in parameters DEFAULT_TILE_SIZE: Tuple[int, int] = (256, 256) #: Maximum size to use when lazy loading metadata (less is faster but less accurate) LAZY_LOADING_MAX_SHAPE: Tuple[int, int] = (1024, 1024) #: Compression level of output PNGs, from 0-9 PNG_COMPRESS_LEVEL: int = 1 #: Timeout in seconds for database connections DB_CONNECTION_TIMEOUT: int = 10 #: Path where cached remote SQLite databases are stored (when using sqlite-remote provider) REMOTE_DB_CACHE_DIR: str = os.path.join(tempfile.gettempdir(), 'terracotta') #: Time-to-live of remote database cache in seconds REMOTE_DB_CACHE_TTL: int = 10 * 60 # 10 min #: Resampling method to use when reading reprojected data RESAMPLING_METHOD: str = 'average' #: Resampling method to use when reprojecting data to Web Mercator REPROJECTION_METHOD: str = 'linear' #: CORS allowed origins for metadata endpoint ALLOWED_ORIGINS_METADATA: List[str] = ['*'] #: CORS allowed origins for tiles endpoints ALLOWED_ORIGINS_TILES: List[str] = []
AVAILABLE_SETTINGS: Tuple[str, ...] = tuple(TerracottaSettings._fields) def _is_writable(path: str) -> bool: return os.access(os.path.dirname(path) or os.getcwd(), os.W_OK) class SettingSchema(Schema): """Schema used to create and validate TerracottaSettings objects""" DRIVER_PATH = fields.String() DRIVER_PROVIDER = fields.String(allow_none=True) DEBUG = fields.Boolean() FLASK_PROFILE = fields.Boolean() XRAY_PROFILE = fields.Boolean() LOGLEVEL = fields.String( validate=validate.OneOf(['debug', 'info', 'warning', 'error', 'critical']) ) RASTER_CACHE_SIZE = fields.Integer(validate=validate.Range(min=0)) RASTER_CACHE_COMPRESS_LEVEL = fields.Integer(validate=validate.Range(min=0, max=9)) DEFAULT_TILE_SIZE = fields.List(fields.Integer(), validate=validate.Length(equal=2)) LAZY_LOADING_MAX_SHAPE = fields.List( fields.Integer(validate=validate.Range(min=0)), validate=validate.Length(equal=2) ) PNG_COMPRESS_LEVEL = fields.Integer(validate=validate.Range(min=0, max=9)) DB_CONNECTION_TIMEOUT = fields.Integer(validate=validate.Range(min=0)) REMOTE_DB_CACHE_DIR = fields.String(validate=_is_writable) REMOTE_DB_CACHE_TTL = fields.Integer(validate=validate.Range(min=0)) RESAMPLING_METHOD = fields.String( validate=validate.OneOf(['nearest', 'linear', 'cubic', 'average']) ) REPROJECTION_METHOD = fields.String( validate=validate.OneOf(['nearest', 'linear', 'cubic', 'average']) ) ALLOWED_ORIGINS_METADATA = fields.List(fields.String()) ALLOWED_ORIGINS_TILES = fields.List(fields.String()) @pre_load def decode_lists(self, data: Dict[str, Any], **kwargs: Any) -> Dict[str, Any]: for var in ('DEFAULT_TILE_SIZE', 'LAZY_LOADING_MAX_SHAPE', 'ALLOWED_ORIGINS_METADATA', 'ALLOWED_ORIGINS_TILES'): val = data.get(var) if val and isinstance(val, str): try: data[var] = json.loads(val) except json.decoder.JSONDecodeError as exc: raise ValidationError( f'Could not parse value for key {var} as JSON: "{val}"' ) from exc return data @post_load def make_settings(self, data: Dict[str, Any], **kwargs: Any) -> TerracottaSettings: # encode tuples for var in ('DEFAULT_TILE_SIZE', 'LAZY_LOADING_MAX_SHAPE', 'ALLOWED_ORIGINS_METADATA', 'ALLOWED_ORIGINS_TILES'): val = data.get(var) if val: data[var] = tuple(val) return TerracottaSettings(**data) def parse_config(config: Mapping[str, Any] = None) -> TerracottaSettings: """Parse given config dict and return new TerracottaSettings object""" config_dict = dict(config or {}) for setting in AVAILABLE_SETTINGS: env_setting = f'TC_{setting}' if setting not in config_dict and env_setting in os.environ: config_dict[setting] = os.environ[env_setting] schema = SettingSchema() try: new_settings = schema.load(config_dict) except ValidationError as exc: raise ValueError('Could not parse configuration') from exc return new_settings