afscgap.flat_cursor

  1"""
  2Interfaces for cursor objects which iterate over records from prejoined flat records.
  3
  4Interfaces for cursor objects which iterate over records from prejoined flat records, either those
  5appearing in the underlying unjoined dataset or zero catch inferred records.
  6
  7(c) 2025 Regents of University of California / The Eric and Wendy Schmidt Center
  8for Data Science and the Environment at UC Berkeley.
  9
 10This file is part of afscgap released under the BSD 3-Clause License. See
 11LICENSE.md.
 12"""
 13import queue
 14import typing
 15
 16import afscgap.cursor
 17import afscgap.model
 18
 19from afscgap.flat_model import RECORDS
 20from afscgap.typesdef import OPT_INT
 21
 22
 23class FlatCursor(afscgap.cursor.Cursor):
 24    """Cursor over flat avro "prejoined" results."""
 25
 26    def __init__(self, records: RECORDS):
 27        """Create a new prejoined cursor.
 28
 29        Create a new cursor over prejoined flat files which have already been parsed or which are
 30        emitted prejoined.
 31
 32        Args:
 33            records: The inner iterator that should be decorated by this cursor.
 34        """
 35        self._records = records
 36        self._records_iter = iter(self._records)
 37
 38    def get_limit(self) -> OPT_INT:
 39        """Get the overall limit.
 40
 41        Returns:
 42            The maximum number of records to return.
 43        """
 44        return None
 45
 46    def get_filtering_incomplete(self) -> bool:
 47        """Determine if this cursor is silently filtering incomplete records.
 48
 49        Returns:
 50            Flag indicating if incomplete records should be silently filtered.
 51            If true, they will not be returned during iteration and placed in
 52            the queue at get_invalid(). If false, they will be returned and
 53            those incomplete records' get_complete() will return false.
 54        """
 55        return False
 56
 57    def get_invalid(self) -> 'queue.Queue[dict]':
 58        """Get a queue of invalid / incomplete records found so far.
 59
 60        Returns:
 61            Queue with dictionaries containing the raw data returned from the
 62            remote that did not have valid values for all required fields. Note
 63            that this will include incomplete records as well if
 64            get_filtering_incomplete() is true and will not contain incomplete
 65            records otherwise.
 66        """
 67        return queue.Queue()
 68
 69    def to_dicts(self) -> typing.Iterable[dict]:
 70        """Create an iterator which converts Records to dicts.
 71
 72        Returns:
 73            Iterator which returns dictionaries instead of Record objects but
 74            has otherwise the same beahavior as iterating in this Cursor
 75            directly.
 76        """
 77        return map(lambda x: x.to_dict(), self)
 78
 79    def get_next(self) -> typing.Optional[afscgap.model.Record]:
 80        """Get the next value for this Cursor.
 81
 82        Returns:
 83            The next value waiting if cached in the cursor's results queue or
 84            as just retrieved from a new page gathered by HTTP request. Will
 85            return None if no remain.
 86        """
 87        try:
 88            return next(self._records_iter)
 89        except StopIteration:
 90            return None
 91
 92
 93class CompleteCursor(afscgap.cursor.Cursor):
 94    """Cursor decorator which only yields complete records."""
 95
 96    def __init__(self, inner: afscgap.cursor.Cursor):
 97        """Create a new decorator for another cursor which filters for complete records.
 98
 99        Args:
100            inner: The cursor to decorate.
101        """
102        self._inner = inner
103        self._invalid: queue.Queue[dict] = queue.Queue()
104
105    def get_limit(self) -> OPT_INT:
106        """Get the overall limit.
107
108        Returns:
109            The maximum number of records to return.
110        """
111        return self._inner.get_limit()
112
113    def get_filtering_incomplete(self) -> bool:
114        """Determine if this cursor is silently filtering incomplete records.
115
116        Returns:
117            Flag indicating if incomplete records should be silently filtered.
118            If true, they will not be returned during iteration and placed in
119            the queue at get_invalid(). If false, they will be returned and
120            those incomplete records' get_complete() will return false.
121        """
122        return True
123
124    def to_dicts(self) -> typing.Iterable[dict]:
125        """Create an iterator which converts Records to dicts.
126
127        Returns:
128            Iterator which returns dictionaries instead of Record objects but
129            has otherwise the same beahavior as iterating in this Cursor
130            directly.
131        """
132        return map(lambda x: x.to_dict(), self)
133
134    def get_next(self) -> typing.Optional[afscgap.model.Record]:
135        """Get the next value for this Cursor.
136
137        Returns:
138            The next value waiting if cached in the cursor's results queue or
139            as just retrieved from a new page gathered by HTTP request. Will
140            return None if no remain.
141        """
142        done = False
143
144        next_candidate = None
145
146        while not done:
147            next_candidate = self._inner.get_next()
148
149            if next_candidate is None:
150                return None
151
152            is_complete = next_candidate.is_complete()
153
154            if not is_complete:
155                next_candidate_cast: afscgap.flat_model.FlatRecord = next_candidate  # type: ignore
156                self._invalid.put(next_candidate_cast.get_inner())
157
158            done = is_complete
159
160        return next_candidate
161
162
163class LimitCursor(afscgap.cursor.Cursor):
164    """Create a decorator which limits cursor iteration to a certain count of returned records."""
165
166    def __init__(self, inner: afscgap.cursor.Cursor, limit: int):
167        """Create a new limit decorator around an existing cursor.
168
169        Create a new limit decorator around an existing cursor which terminates iteration either
170        after a limit is reached or no additional records are available.
171
172        Args:
173            inner: The cursor to decorate.
174            limit: The integer count of records after which iteration will be terminated.
175        """
176        self._inner = inner
177        self._limit = limit
178        self._remaining = limit
179
180    def get_limit(self) -> OPT_INT:
181        """Get the overall limit.
182
183        Returns:
184            The maximum number of records to return.
185        """
186        return self._limit
187
188    def get_filtering_incomplete(self) -> bool:
189        """Determine if this cursor is silently filtering incomplete records.
190
191        Returns:
192            Flag indicating if incomplete records should be silently filtered.
193            If true, they will not be returned during iteration and placed in
194            the queue at get_invalid(). If false, they will be returned and
195            those incomplete records' get_complete() will return false.
196        """
197        return self._inner.get_filtering_incomplete()
198
199    def to_dicts(self) -> typing.Iterable[dict]:
200        """Create an iterator which converts Records to dicts.
201
202        Returns:
203            Iterator which returns dictionaries instead of Record objects but
204            has otherwise the same beahavior as iterating in this Cursor
205            directly.
206        """
207        return map(lambda x: x.to_dict(), self)
208
209    def get_next(self) -> typing.Optional[afscgap.model.Record]:
210        """Get the next value for this Cursor.
211
212        Returns:
213            The next value waiting if cached in the cursor's results queue or
214            as just retrieved from a new page gathered by HTTP request. Will
215            return None if no remain.
216        """
217        if self._remaining == 0:
218            return None
219
220        next_candidate = self._inner.get_next()
221
222        if next_candidate is None:
223            return None
224        else:
225            self._remaining -= 1
226            return next_candidate