Source code for harvester.model

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Optional

from vs_common.model import FilesystemConfig, FilesystemType, LocalFilesystemConfig


class ResourceType(str, Enum):
    STACAPI = "stacapi"
    STACCatalog = "staccatalog"
    FileMatcher = "filematcher"
    OADS = "oads"
    OpenSearch = "opensearch"


class PostprocessorType(str, Enum):
    builtin = "builtin"
    external = "external"


class OutputType(str, Enum):
    console = "console"
    queue = "queue"


@dataclass
class TimeConfig:
    begin: str = datetime.now().isoformat()
    end: str = datetime.now().isoformat()


@dataclass
class QueryConfig:
    time: TimeConfig
    bbox: str
    collection: Optional[str] = None
    extra_params: Dict[str, str] = field(default_factory=dict)


@dataclass
class STACAPIConfig:
    url: str
    query: QueryConfig


@dataclass
class JSONConfig:
    property_mapping: Dict[str, str]


@dataclass
class ATOMXMLConfig:
    pass


@dataclass
class FormatConfig:
    type: str
    json: Optional[JSONConfig] = None
    atom_xml: Optional[ATOMXMLConfig] = None


@dataclass
class STACCatalogConfig:
    root_path: str
    filesystem: str
    collection_id: Optional[str] = None
    deduplicate: bool = False


@dataclass
class FileMatcherConfig:
    root_path: str
    filesystem: str
    asset_regex_map: Dict[str, str]
    id_regex: str
    datetime_regex: str


@dataclass
class OADSConfig:
    url: str
    use_oads_ext: bool = False


@dataclass
class OpenSearchConfig:
    url: str
    query: QueryConfig
    format: FormatConfig


@dataclass
class ResourceConfig:
    type: ResourceType
    stacapi: Optional[STACAPIConfig] = None
    staccatalog: Optional[STACCatalogConfig] = None
    filematcher: Optional[FileMatcherConfig] = None
    oads: Optional[OADSConfig] = None
    opensearch: Optional[OpenSearchConfig] = None


@dataclass
class PostprocessorConfig:
    type: PostprocessorType
    process: str
    kwargs: Dict[str, Any] = field(default_factory=dict)


@dataclass
class FilterConfig:
    context: Dict = field(default_factory=dict)
    expression: Dict = field(default_factory=dict)


@dataclass
class HarvesterConfig:
    resource: ResourceConfig
    filter: Optional[FilterConfig] = None
    output: OutputType = OutputType.queue
    queue: Optional[str] = None
    postprocessors: Optional[List[PostprocessorConfig]] = None


@dataclass
class RedisConfig:
    host: str = "vs-redis-master"
    port: int = 6379


def default_filesystem():
    return {
        "local": FilesystemConfig(FilesystemType.file, local=LocalFilesystemConfig())
    }


@dataclass
class HarvesterAppConfig:
    harvesters: Dict[str, HarvesterConfig] = field(default_factory=dict)
    redis: RedisConfig = RedisConfig()
    filesystems: Dict[str, FilesystemConfig] = field(default_factory=default_filesystem)