Skip to content

Record Collection

Bases: DataFrame

A RecordCollection is an extention of a Pandas DataFrame.

It can be used in precisely the same way as a Pandas DataFrame but has a few extra useful constructors.

Parameters:

Name Type Description Default
_scope CollectionScope | None

params for generating api query for pagination

 required
_page int | None

if generated from an API query, the current page of the query.

 required
Source code in tz/client/record.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 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
 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
110
class RecordCollection(pd.DataFrame):
    """A RecordCollection is an extention of a Pandas DataFrame.

    It can be used in precisely the same way as a Pandas DataFrame
    but has a few extra useful constructors.

    Args:
        _scope (schemas.CollectionScope | None): params for generating api query for pagination
        _page (int | None): if generated from an API query, the current page of the query.
    """

    _scope: Optional[schemas.CollectionScope] = None
    _page: Optional[int] = None

    @property
    def _constructor(self):
        return RecordCollection

    @property
    def _constructor_sliced(self):
        return RecordCollectionRow

    @classmethod
    def search(
        cls,
        node_id: str | None = None,
        public: bool = True,
        valid_timestamp_start: datetime | None = None,
        valid_timestamp_end: datetime | None = None,
        provenance_slug: list[str] | None = None,
        datum_type: list[str] | None = None,
        datum_detail: list[str] | None = None,
        node_type: list[str] | None = None,
        technology: str | None = None,
        limit: int | None = 10,
        page: int | None = 0,
    ):
        """Instantiate a RecordCollection from a node or technology.

        Args:
            node_id (str): The id of the node to retieve records for.
            public (bool): Whether to include public records.
            valid_timestamp_start (datetime): Start timestamp of record validity.
            valid_timestamp_end (datetime): End timestamp of record validity.
            provenance_slug (list[str]): Provenance of the record (e.g. 'Copernicus-Landuse').
            datum_type (list[str]): Datum type of the record (e.g. 'Landuse').
            datum_detail (list[str]): Datum detail of the record.
            node_type (list[str]): Node type of the record (e.g. 'country').
            technology: Technology slug of the record (e.g. 'coal').
            limit (int): The maximum number of search results to return.
            page (int): The page number of search results to return.
        Returns:
            RecordCollection: A pandas-dataframe extension for TZ records.
        """
        records = api.records.get(
            node_id=node_id,
            public=public,
            valid_timestamp_start=valid_timestamp_start,
            valid_timestamp_end=valid_timestamp_end,
            provenance_slug=provenance_slug,
            datum_type=datum_type,
            datum_detail=datum_detail,
            node_type=node_type,
            technology=technology,
            limit=limit,
            page=page,
        )

        obj = cls.from_tz_records(records)  # type: ignore[arg-type]
        obj._scope = schemas.CollectionScope(node_id=node_id)
        obj._page = 0
        return obj

    @classmethod
    def from_tz_records(cls, records: List[Record]):
        """Instiate an RecordCollection from a list of Records."""
        # pd.DataFrame.from_records
        return cls.from_records([record.model_dump() for record in records])

    def next_page(self):
        """Paginate through records. The Record collection must have a `_scope`.

        Returns the next page of records and concatenates them in-place to the current collection.
        """
        if not self._scope:
            raise ValueError("Cant iterate an unscoped RecordCollection")
        new_collection = self.__class__.from_tz_records(
            api.records.get(node_id=self._scope.node_id, page=self._page + 1)
        )
        self._page += 1

        self.__dict__.update(pd.concat([self, new_collection], ignore_index=True).__dict__)
        return len(new_collection)

    def to_tz_records(self):
        """Instantiate a list of Records from an RecordCollection."""
        return [Record(**row) for idx, row in self.iterrows()]

from_tz_records(records) classmethod

Instiate an RecordCollection from a list of Records.

Source code in tz/client/record.py 
87
88
89
90
91
@classmethod
def from_tz_records(cls, records: List[Record]):
    """Instiate an RecordCollection from a list of Records."""
    # pd.DataFrame.from_records
    return cls.from_records([record.model_dump() for record in records])

next_page()

Paginate through records. The Record collection must have a _scope.

Returns the next page of records and concatenates them in-place to the current collection.

Source code in tz/client/record.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def next_page(self):
    """Paginate through records. The Record collection must have a `_scope`.

    Returns the next page of records and concatenates them in-place to the current collection.
    """
    if not self._scope:
        raise ValueError("Cant iterate an unscoped RecordCollection")
    new_collection = self.__class__.from_tz_records(
        api.records.get(node_id=self._scope.node_id, page=self._page + 1)
    )
    self._page += 1

    self.__dict__.update(pd.concat([self, new_collection], ignore_index=True).__dict__)
    return len(new_collection)

search(node_id=None, public=True, valid_timestamp_start=None, valid_timestamp_end=None, provenance_slug=None, datum_type=None, datum_detail=None, node_type=None, technology=None, limit=10, page=0) classmethod

Instantiate a RecordCollection from a node or technology.

Parameters:

Name Type Description Default
node_id str

The id of the node to retieve records for.

 None
public bool

Whether to include public records.

 True
valid_timestamp_start datetime

Start timestamp of record validity.

 None
valid_timestamp_end datetime

End timestamp of record validity.

 None
provenance_slug list[str]

Provenance of the record (e.g. 'Copernicus-Landuse').

 None
datum_type list[str]

Datum type of the record (e.g. 'Landuse').

 None
datum_detail list[str]

Datum detail of the record.

 None
node_type list[str]

Node type of the record (e.g. 'country').

 None
technology str | None

Technology slug of the record (e.g. 'coal').

 None
limit int

The maximum number of search results to return.

 10
page int

The page number of search results to return.

 0

Returns: RecordCollection: A pandas-dataframe extension for TZ records.

Source code in tz/client/record.py 
36
37
38
39
40
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@classmethod
def search(
    cls,
    node_id: str | None = None,
    public: bool = True,
    valid_timestamp_start: datetime | None = None,
    valid_timestamp_end: datetime | None = None,
    provenance_slug: list[str] | None = None,
    datum_type: list[str] | None = None,
    datum_detail: list[str] | None = None,
    node_type: list[str] | None = None,
    technology: str | None = None,
    limit: int | None = 10,
    page: int | None = 0,
):
    """Instantiate a RecordCollection from a node or technology.

    Args:
        node_id (str): The id of the node to retieve records for.
        public (bool): Whether to include public records.
        valid_timestamp_start (datetime): Start timestamp of record validity.
        valid_timestamp_end (datetime): End timestamp of record validity.
        provenance_slug (list[str]): Provenance of the record (e.g. 'Copernicus-Landuse').
        datum_type (list[str]): Datum type of the record (e.g. 'Landuse').
        datum_detail (list[str]): Datum detail of the record.
        node_type (list[str]): Node type of the record (e.g. 'country').
        technology: Technology slug of the record (e.g. 'coal').
        limit (int): The maximum number of search results to return.
        page (int): The page number of search results to return.
    Returns:
        RecordCollection: A pandas-dataframe extension for TZ records.
    """
    records = api.records.get(
        node_id=node_id,
        public=public,
        valid_timestamp_start=valid_timestamp_start,
        valid_timestamp_end=valid_timestamp_end,
        provenance_slug=provenance_slug,
        datum_type=datum_type,
        datum_detail=datum_detail,
        node_type=node_type,
        technology=technology,
        limit=limit,
        page=page,
    )

    obj = cls.from_tz_records(records)  # type: ignore[arg-type]
    obj._scope = schemas.CollectionScope(node_id=node_id)
    obj._page = 0
    return obj

to_tz_records()

Instantiate a list of Records from an RecordCollection.

Source code in tz/client/record.py 
108
109
110
def to_tz_records(self):
    """Instantiate a list of Records from an RecordCollection."""
    return [Record(**row) for idx, row in self.iterrows()]