Skip to content

Ceda vocabulary

CEDAVocabularyExtract

Bases: ExtractionMethod

Validates and sorts properties into vocabs and generates the general vocab for specified properties.

Method name: ceda_vocabulary

Example configuration

.. code-block:: yaml

  • method: ceda_vocabulary inputs: url: vocab.ceda.ac.uk namespace: cmip6 strict: False terms:
    • start_time
    • model
Source code in extraction_methods/plugins/ceda_vocabulary.py 
 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
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class CEDAVocabularyExtract(ExtractionMethod):
    """
    Validates and sorts properties into vocabs and generates
    the `general` vocab for specified properties.

    **Method name:** ``ceda_vocabulary``

    Example configuration:
        .. code-block:: yaml

        - method: ceda_vocabulary
          inputs:
            url: vocab.ceda.ac.uk
            namespace: cmip6
            strict: False
            terms:
              - start_time
              - model
    """

    input_class = CEDAVocabularyInput

    @update_input
    def run(self, body: dict[str, Any]) -> dict[str, Any]:

        properties = body

        if "unspecified_vocab" in body:
            properties = body["unspecified_vocab"]

        req_data = {
            "namespace": self.input.namespace,
            "terms": self.input.terms,
            "properties": properties,
            "strict": self.input.strict,
        }

        response = httpx.post(
            self.input.url,
            json=req_data,
            timeout=self.input.request_timeout,
        )

        if response.status_code != 200:
            raise Exception(
                f"Bad response from vocab server: {response.status_code}, reason: {response.text}"
            )

        json_response = response.json()

        if json_response["error"]:
            raise Exception(f"Vocab request failed, reason: {json_response['text']}")

        body = body | json_response["result"]

        if "vocabs" in body:
            body["vocabs"].append(self.input.namespace)

        else:
            body["vocabs"] = self.input.namespace

        return body

CEDAVocabularyInput

Bases: Input

Model for CEDA Vocab Method Input.

Parameters:

Name Type Description Default
url str

URL of vocabulary server.

 required
namespace str

Namespace for vocab terms.

 required
strict bool

True if values should be validated.

 False
terms list[str]

terms to be validated.

 []
request_timeout int

request time out.

 15
Source code in extraction_methods/plugins/ceda_vocabulary.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class CEDAVocabularyInput(Input):
    """
    Model for CEDA Vocab Method Input.
    """

    url: str = Field(
        description="URL of vocabulary server.",
    )
    namespace: str = Field(
        description="Namespace for vocab terms.",
    )
    strict: bool = Field(
        default=False,
        description="True if values should be validated.",
    )
    terms: list[str] = Field(
        default=[],
        description="terms to be validated.",
    )
    request_timeout: int = Field(
        default=15,
        description="request time out.",
    )