lms.model.base

  1import typing
  2
  3import edq.util.json
  4import edq.util.serial
  5import edq.util.time
  6
  7import lms.model.constants
  8import lms.util.string
  9
 10TEXT_SEPARATOR: str = ': '
 11TEXT_EMPTY_VALUE: str = ''
 12
 13T = typing.TypeVar('T', bound = 'BaseType')
 14
 15class BaseType(edq.util.serial.DictConverter):
 16    """
 17    The base class for all core LMS types.
 18    This class ensures that all children have the core functionality necessary for this package.
 19
 20    The typical structure of types in this package is that types in the model package extend this class.
 21    Then, backends may declare their own types that extend the other classes from the model package.
 22    For example: lms.model.base.BaseType -> lms.model.assignments.Assignment -> lms.backend.canvas.model.assignments.Assignment
 23
 24    General (but less efficient) implementations of core functions will be provided.
 25    """
 26
 27    CORE_FIELDS: typing.List[str] = []
 28    """
 29    The common fields shared across backends for this type that are used for comparison and other operations.
 30    Child classes should set this to define how comparisons are made.
 31    """
 32
 33    INT_COMPARISON_FIELDS: typing.Set[str] = {'id'}
 34    """
 35    Fields that should be compared like ints (even if they are strings).
 36    By default, this set will include 'id'.
 37    """
 38
 39    def __init__(self,
 40            **kwargs: typing.Any) -> None:
 41        self.extra_fields: typing.Dict[str, typing.Any] = kwargs.copy()
 42        """ Additional fields not common to all backends or explicitly used by the creating child backend. """
 43
 44    def __eq__(self, other: object) -> bool:
 45        if (not isinstance(other, BaseType)):
 46            return False
 47
 48        # Check the specified fields only.
 49        for field_name in self.CORE_FIELDS:
 50            if (not hasattr(other, field_name)):
 51                return False
 52
 53            value_self = getattr(self, field_name)
 54            value_other = getattr(other, field_name)
 55
 56            if (field_name in self.INT_COMPARISON_FIELDS):
 57                comparison = lms.util.string.compare_maybe_ints(value_self, value_other)
 58                if (comparison != 0):
 59                    return False
 60            elif (value_self != value_other):
 61                return False
 62
 63        return True
 64
 65    def __hash__(self) -> int:
 66        values = tuple(getattr(self, field_name) for field_name in self.CORE_FIELDS)
 67        return hash(values)
 68
 69    def __lt__(self, other: 'BaseType') -> bool:  # type: ignore[override]
 70        if (not isinstance(other, BaseType)):
 71            return False
 72
 73        # Check the specified fields only.
 74        for field_name in self.CORE_FIELDS:
 75            if (not hasattr(other, field_name)):
 76                return False
 77
 78            value_self = getattr(self, field_name)
 79            value_other = getattr(other, field_name)
 80
 81            if (field_name in self.INT_COMPARISON_FIELDS):
 82                comparison = lms.util.string.compare_maybe_ints(value_self, value_other)
 83                if (comparison == 0):
 84                    continue
 85
 86                return (comparison < 0)
 87
 88            if (value_self == value_other):
 89                continue
 90
 91            return bool(value_self < value_other)
 92
 93        return False
 94
 95    def as_text_rows(self,
 96            skip_headers: bool = False,
 97            pretty_headers: bool = False,
 98            **kwargs: typing.Any) -> typing.List[str]:
 99        """
100        Create a representation of this object in the "text" style of this project meant for display.
101        A list of rows will be returned.
102        """
103
104        rows = []
105
106        kwargs['pretty_timestamps'] = True
107
108        for (field_name, row) in self._get_fields(**kwargs).items():
109            if (not skip_headers):
110                header = field_name
111                if (pretty_headers):
112                    header = header.replace('_', ' ').title()
113
114                row = f"{header}{TEXT_SEPARATOR}{row}"
115
116            rows.append(row)
117
118        return rows
119
120    def get_headers(self,
121            pretty_headers: bool = False,
122            **kwargs: typing.Any) -> typing.List[str]:
123        """
124        Get a list of headers to label the values represented by this object meant for display.
125        This method is a companion to as_table_rows(),
126        given the same options these two methods will produce rows with the same length and ordering.
127        """
128
129        headers = []
130
131        for field_name in self._get_fields(**kwargs):
132            header = field_name
133            if (pretty_headers):
134                header = header.replace('_', ' ').title()
135
136            headers.append(header)
137
138        return headers
139
140    def as_table_rows(self,
141            **kwargs: typing.Any) -> typing.List[typing.List[str]]:
142        """
143        Get a list of the values by this object meant for display.
144        This method is a companion to get_headers(),
145        given the same options these two methods will produce rows with the same length and ordering.
146
147        Note that the default implementation for this method always return a single row,
148        but children may override and return multiple rows per object.
149        """
150
151        return [list(self._get_fields(**kwargs).values())]
152
153    def as_json_dict(self,
154            **kwargs: typing.Any) -> typing.Dict[str, typing.Any]:
155        """
156        Get a dict representation of this object meant for display as JSON.
157        (Note that we are not returning JSON, just a dict that is ready to be converted to JSON.)
158        Calling this method differs from passing this object to json.dumps() (or any sibling),
159        because this method may not include all fields, may flatten or alter fields, and will order fields differently.
160        """
161
162        return {
163            field_name: self._get_field_value(field_name)
164            for field_name
165            in self._get_fields(**kwargs)
166        }
167
168    def _get_fields(self,
169            include_extra_fields: bool = False,
170            **kwargs: typing.Any) -> typing.Dict[str, str]:
171        """
172        Get a dictionary representing the "target" fields of this object meant for display.
173        Keys (field names) will not be modified, but values will be sent to self._value_to_text().
174        Keys are placed in the dictionary in a consistent ordering.
175        """
176
177        field_names = self.CORE_FIELDS.copy()
178
179        # Append any extra fields after the core fields.
180        if (include_extra_fields):
181            # First, include any fields that are not in self.extra_fields.
182            for extra_name in (list(vars(self).keys()) + list(self.extra_fields.keys())):
183                if (extra_name == 'extra_fields'):
184                    continue
185
186                if (extra_name not in field_names):
187                    field_names.append(extra_name)
188
189        fields = {}
190        for field_name in field_names:
191            fields[field_name] = self._value_to_text(self._get_field_value(field_name), **kwargs)
192
193        return fields
194
195    def _get_field_value(self, name: str, default: typing.Any = None) -> typing.Any:
196        """
197        Get the value for a field.
198        This is similar to `getattr(self, name, default)`,
199        but this will also check `extra_fields` if the field is not found at the top level.
200        """
201
202        if (hasattr(self, name)):
203            return getattr(self, name)
204
205        if (name in self.extra_fields):
206            return self.extra_fields[name]
207
208        return default
209
210    def _value_to_text(self,
211            value: typing.Any,
212            indent: typing.Union[int, None] = None,
213            pretty_timestamps: bool = False,
214            **kwargs: typing.Any) -> str:
215        """
216        Convert some arbitrary value (usually found within a BaseType) to a string.
217        None values will be returned as `TEXT_EMPTY_VALUE`.
218        """
219
220        if (value is None):
221            return TEXT_EMPTY_VALUE
222
223        if (hasattr(value, '_to_text')):
224            return str(value._to_text())
225
226        if (pretty_timestamps and isinstance(value, edq.util.time.Timestamp)):
227            return value.pretty(short = True)
228
229        if (isinstance(value, (edq.util.serial.PODSerializer, dict, list, tuple))):
230            return str(edq.util.json.dumps(value, indent = indent))
231
232        return str(value)
233
234    @classmethod
235    def from_json_dict(cls: typing.Type[T],
236            data: typing.Dict[str, typing.Any],
237            **kwargs: typing.Any) -> T:
238        """
239        Create an object from a dict that can be used for JSON.
240        This is the inverse of as_json_dict().
241        """
242
243        return cls.from_dict(data)
244
245def base_list_to_output_format(
246        values: typing.Sequence[BaseType],
247        output_format: lms.model.constants.OutputFormat,
248        sort: bool = True,
249        skip_headers: bool = False,
250        pretty_headers: bool = False,
251        include_extra_fields: bool = False,
252        **kwargs: typing.Any) -> str:
253    """
254    Convert a list of base types to a string representation.
255    The returned string will not include a trailing newline.
256
257    The given list may be modified by this call.
258    """
259
260    values = list(values)
261
262    if (sort):
263        values.sort()
264
265    output = ''
266
267    if (output_format == lms.model.constants.OutputFormat.JSON):
268        output = base_list_to_json(values,
269                include_extra_fields = include_extra_fields,
270                **kwargs)
271    elif (output_format == lms.model.constants.OutputFormat.TABLE):
272        output = base_list_to_table(values,
273                skip_headers = skip_headers, pretty_headers = pretty_headers,
274                include_extra_fields = include_extra_fields,
275                **kwargs)
276    elif (output_format == lms.model.constants.OutputFormat.TEXT):
277        output = base_list_to_text(values,
278                skip_headers = skip_headers, pretty_headers = pretty_headers,
279                include_extra_fields = include_extra_fields,
280                **kwargs)
281    else:
282        raise ValueError(f"Unknown output format: '{output_format}'.")
283
284    return output
285
286def base_list_to_json(values: typing.Sequence[BaseType],
287        indent: int = 4,
288        extract_single_list: bool = False,
289        **kwargs: typing.Any) -> str:
290    """ Convert a list of base types to a JSON string representation. """
291
292    output_values = [value.as_json_dict(**kwargs) for value in values]
293    if (extract_single_list and (len(output_values) == 1)):
294        output_values = output_values[0]  # type: ignore[assignment]
295
296    return str(edq.util.json.dumps(output_values, indent = indent, sort_keys = False))
297
298def base_list_to_table(values: typing.Sequence[BaseType],
299        skip_headers: bool = False,
300        delim: str = "\t",
301        **kwargs: typing.Any) -> str:
302    """ Convert a list of base types to a table string representation. """
303
304    rows = []
305
306    if ((len(values) > 0) and (not skip_headers)):
307        rows.append(values[0].get_headers(**kwargs))
308
309    for value in values:
310        rows += value.as_table_rows(**kwargs)
311
312    return "\n".join([delim.join(row) for row in rows])
313
314def base_list_to_text(values: typing.Sequence[BaseType],
315        **kwargs: typing.Any) -> str:
316    """ Convert a list of base types to a text string representation. """
317
318    output = []
319
320    for value in values:
321        rows = value.as_text_rows(**kwargs)
322        output.append("\n".join(rows))
323
324    return "\n\n".join(output)