Skip to content

Ceda observation

CEDAObservationExtract

Bases: ExtractionMethod

Returns a ceda observation record for the input_term.

Method name: ceda_observation

Example Configuration

.. code-block:: yaml

- method: ceda_observation
  inputs:
    input_term: $url
Source code in extraction_methods/plugins/ceda_observation.py 
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
class CEDAObservationExtract(ExtractionMethod):
    """
    Returns a ceda observation record for the ``input_term``.

    **Method name:** ``ceda_observation``

    Example Configuration:
        .. code-block:: yaml

            - method: ceda_observation
              inputs:
                input_term: $url
    """

    input_class = CEDAObservationInput

    @update_input
    def run(self, body: dict[str, Any]) -> dict[str, Any]:
        r = httpx.get(self.input.input_term, timeout=self.input.request_timeout)

        if r.status_code == 200:
            response = r.json()
            record_type = response.get("record_type")
            url = response.get("url")

            if record_type == "Dataset" and url:
                body[self.input.output_key] = url.split("/")[-1]

        return body

CEDAObservationInput

Bases: Input

Model for CEDA Observation Method Input.

Parameters:

Name Type Description Default
input_term str

term for method to run on.

 '$uri'
request_timeout int

request time out.

 15
output_key str

key to output to.

 'uuid'
Source code in extraction_methods/plugins/ceda_observation.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class CEDAObservationInput(Input):
    """
    Model for CEDA Observation Method Input.
    """

    input_term: str = Field(
        default="$uri",
        description="term for method to run on.",
    )
    request_timeout: int = Field(
        default=15,
        description="request time out.",
    )
    output_key: str = Field(
        default="uuid",
        description="key to output to.",
    )