baserow.field

  1"""
  2The module contains definitions for the values of table fields that do not
  3directly translate into built-in types.
  4"""
  5
  6from __future__ import annotations
  7import abc
  8import datetime
  9import enum
 10from io import BufferedReader
 11from pathlib import Path
 12from typing import TYPE_CHECKING, Generic, Optional, TypeVar, Union
 13
 14from pydantic import BaseModel, ConfigDict, Field, RootModel, model_serializer, model_validator
 15
 16from baserow.client import GlobalClient
 17from baserow.color import ColorSequence
 18from baserow.error import PydanticGenericMetadataError
 19from baserow.field_config import CreatedByFieldConfig, CreatedOnFieldConfig, FieldConfigType, FileFieldConfig, LastModifiedByFieldConfig, LastModifiedFieldConfig, MultipleCollaboratorsFieldConfig, MultipleSelectFieldConfig, SelectEntryConfig, SingleSelectFieldConfig
 20from baserow.file import File
 21
 22if TYPE_CHECKING:
 23    from baserow.client import Client
 24
 25
 26class FieldType(str, enum.Enum):
 27    """The various types that Baserow fields can have."""
 28    TEXT = "text"
 29    NUMBER = "number"
 30    LONG_TEXT = "long_text"
 31    LINK_ROW = "link_row"
 32    BOOLEAN = "boolean"
 33    DATE = "date"
 34    RATING = "rating"
 35    LAST_MODIFIED = "last_modified"
 36    LAST_MODIFIED_BY = "last_modified_by"
 37    CREATED_ON = "created_on"
 38    CREATED_BY = "created_by"
 39    DURATION = "duration"
 40    URL = "url"
 41    EMAIL = "email"
 42    FILE = "file"
 43    SINGLE_SELECT = "single_select"
 44    MULTIPLE_SELECT = "multiple_select"
 45    PHONE_NUMBER = "phone_number"
 46    FORMULA = "formula"
 47    ROLLUP = "rollup"
 48    LOOKUP = "lookup"
 49    MULTIPLE_COLLABORATORS = "multiple_collaborators"
 50    UUID = "uuid"
 51    AUTONUMBER = "autonumber"
 52    PASSWORD = "password"
 53
 54
 55class BaserowField(BaseModel, abc.ABC):
 56    """
 57    Abstract base class for all Baserow fields that are not covered by the
 58    built-in types.
 59
 60    This class also handles tracking changes, which are initially local and only
 61    applied to Baserow when Table.update() is called. For example, the method
 62    TableLinkField.append() adds a new link to a record, but this change is only
 63    written to Baserow when Table.update() is invoked. Such actions can be
 64    registered with BaserowField.register_pending_change(). If an object with
 65    pending changes is deleted (__del__), a corresponding warning is issued.
 66    """
 67
 68    @classmethod
 69    @abc.abstractmethod
 70    def default_config(cls) -> FieldConfigType:
 71        """Returns the default field config for a given field type."""
 72
 73    @classmethod
 74    @abc.abstractmethod
 75    def read_only(cls) -> bool:
 76        """
 77        Read only fields (like modification date or UUID fields) will be ignored
 78        during update calls like `Table.update()`.
 79        """
 80
 81    _pending_changes: list[str] = []
 82    """See documentation of `BaserowField.register_pending_changes()`."""
 83
 84    def register_pending_change(self, description: str):
 85        """
 86        Registers a change to the field, which is currently only stored locally
 87        and must be explicitly applied to Baserow using `Table.update()`.
 88
 89        Args:
 90            description (str): A description of the change in data.
 91        """
 92        self._pending_changes.append(description)
 93
 94    def changes_applied(self):
 95        """
 96        Is called by `Table.update()` after all pending changes have been
 97        written to Baserow.
 98        """
 99        self._pending_changes = []
100
101    def __del__(self):
102        if len(self._pending_changes) != 0:
103            changes = ["- " + change for change in self._pending_changes]
104            print(f"WARNING: THERE ARE STILL PENDING CHANGES IN FIELD {self.__class__.__name__}")  # noqa: F821
105            print("\n".join(changes))
106            print(
107                "It looks like `Table.update()` was not called to apply these changes to Baserow.",
108            )
109
110
111class User(BaseModel):
112    """
113    A table field that contains one Baserow system user.
114    """
115    user_id: Optional[int] = Field(alias=str("id"))
116    name: Optional[str] = Field(alias=str("name"))
117
118
119class CreatedByField(User, BaserowField):
120    """
121    Field tracking the user who created an entry.
122    """
123    @classmethod
124    def default_config(cls) -> FieldConfigType:
125        return CreatedByFieldConfig()
126
127    @classmethod
128    def read_only(cls) -> bool:
129        return True
130
131
132class LastModifiedByField(User, BaserowField):
133    """
134    Field tracking the user who last modified a record.
135    """
136    @classmethod
137    def default_config(cls) -> FieldConfigType:
138        return LastModifiedByFieldConfig()
139
140    @classmethod
141    def read_only(cls) -> bool:
142        return True
143
144
145class CreatedOnField(BaserowField, RootModel[datetime.datetime]):
146    """
147    Field containing the creation timestamp of a row.
148    """
149    root: datetime.datetime
150
151    @classmethod
152    def default_config(cls) -> FieldConfigType:
153        return CreatedOnFieldConfig()
154
155    @classmethod
156    def read_only(cls) -> bool:
157        return True
158
159
160class LastModifiedOnField(BaserowField, RootModel[datetime.datetime]):
161    """
162    Field containing the last modification timestamp of a row.
163    """
164    root: datetime.datetime
165
166    @classmethod
167    def default_config(cls) -> FieldConfigType:
168        return LastModifiedFieldConfig()
169
170    @classmethod
171    def read_only(cls) -> bool:
172        return True
173
174
175class MultipleCollaboratorsField(BaserowField, RootModel[list[User]]):
176    """
177    A table field that contains one or multiple Baserow system user(s).
178    """
179    root: list[User]
180
181    @classmethod
182    def default_config(cls) -> FieldConfigType:
183        return MultipleCollaboratorsFieldConfig()
184
185    @classmethod
186    def read_only(cls) -> bool:
187        return False
188
189
190class FileField(BaserowField, RootModel[list[File]]):
191    """
192    A file field allows you to easily upload one or more files from your device
193    or from a URL.
194    """
195    root: list[File]
196
197    @classmethod
198    def default_config(cls) -> FieldConfigType:
199        return FileFieldConfig()
200
201    @classmethod
202    def read_only(cls) -> bool:
203        return False
204
205    @classmethod
206    async def from_file(
207        cls,
208        file: BufferedReader | str | Path,
209        name: Optional[str] = None,
210        client: Optional[Client] = None,
211    ):
212        """
213        This method takes a local file (either as a `BufferedReader` or as a
214        path) and uploads it to Baserow's media storage, handling the linkage
215        between the field and the stored file. A `FileField` instance containing
216        the uploaded file is returned. Calling this method initiates the file
217        upload, which may take some time depending on the file size.
218
219        Note that this method does not use the client associated with the table.
220
221        ```python
222        # By path.
223        await Book(
224            cover=await FileField.from_file("path/to/image.png"),
225        ).create()
226
227        # With BufferedReader.
228        await Book(
229            cover=await FileField.from_file(open("path/to/image.png", "rb")),
230        ).create()
231        ```
232
233        Args:
234            file (BufferedReader | str | Path): File to be uploaded.
235            name (str | None): Optional human-readable name of the file, as it
236                should be displayed in the Baserow frontend.
237            client (Client | None): Specifies which API client should be used.
238                If none is provided, the `GlobalClient` will be used. Note that
239                this method does not automatically use the client associated
240                with the table.
241        """
242        rsl = cls(root=[])
243        await rsl.append_file(file, name, client, register_pending_change=False)
244        return rsl
245
246    @classmethod
247    async def from_url(
248        cls,
249        url: str,
250        name: Optional[str] = None,
251        client: Optional[Client] = None,
252    ):
253        """
254        This method takes the URL of a publicly accessible file on the internet
255        and uploads it to Baserow's media storage, managing the linkage between
256        the field and the stored file. Calling this method initiates the file
257        upload, which may take some time depending on the file size.
258
259        Note that this method does not use the client associated with the table.
260
261        ```python
262        await Book(
263            cover=await FileField.from_url("https://example.com/image.png"),
264        ).create()
265        ```
266
267        Args:
268            url (str): URL of the file to be uploaded.
269            name (str | None): Optional human-readable name of the file, as it
270                should be displayed in the Baserow frontend.
271            client (Client | None): Specifies which API client should be used.
272                If none is provided, the `GlobalClient` will be used. Note that
273                this method does not automatically use the client associated
274                with the table.
275        """
276        rsl = cls(root=[])
277        await rsl.append_file_from_url(url, name, client, register_pending_change=False)
278        return rsl
279
280    async def append_file(
281        self,
282        file: BufferedReader | str | Path,
283        name: Optional[str] = None,
284        client: Optional[Client] = None,
285        register_pending_change: bool = True,
286    ):
287        """
288        This method takes a local file (either as a `BufferedReader` or as a
289        path) and uploads it to Baserow's media storage and adds it to the file
290        field.Calling this method initiates the file upload, which may take some
291        time depending on the file size.
292
293        Further information about uploading and setting files can be found in
294        the documentation of `client.upload_file()`. After this method
295        `Table.update()` must be called manually to apply the changes.
296
297        ```python
298        book = Book.by_id(ROW_ID)
299
300        # By path.
301        await book.cover.append_file("path/to/image.png")
302
303        # With BufferedReader.
304        await book.cover.append_file(open("path/to/image.png", "rb"))
305
306        book.update()
307        ```
308
309        Args:
310            file (BufferedReader | str | Path): File to be uploaded.
311            name (str | None): Optional human-readable name of the file, as it
312                should be displayed in the Baserow frontend.
313            client (Client | None): Specifies which API client should be used.
314                If none is provided, the `GlobalClient` will be used. Note that
315                this method does not automatically use the client associated
316                with the table.
317            register_pending_change (bool): Internal option for the
318                `FileField.from_file()` and `FileField.from_url()` methods. When
319                set to false, tracking of pending changes for this call is
320                disabled. This is only useful in situations where
321                `Table.update()` will be called anyway, such as when the method
322                is invoked within the initialization of a model.
323        """
324        if not isinstance(file, BufferedReader):
325            file = open(file, "rb")
326        if client is None:
327            client = GlobalClient()
328        new_file = await client.upload_file(file)
329        if name is not None:
330            new_file.original_name = name
331        self.root.append(new_file)
332        if register_pending_change:
333            self.register_pending_change(
334                f"file '{new_file.original_name}' added",
335            )
336
337    async def append_file_from_url(
338        self,
339        url: str,
340        name: Optional[str] = None,
341        client: Optional[Client] = None,
342        register_pending_change: bool = True,
343    ):
344        """
345        This method takes the URL of a publicly accessible file on the internet
346        and uploads it to Baserow's media storage and adds it to the file
347        field.Calling this method initiates the file upload, which may take some
348        time depending on the file size.
349
350        Further information about uploading and setting files can be found in
351        the documentation of `client.upload_file()`. After this method
352        `Table.update()` must be called manually to apply the changes.
353
354        ```python
355        book = Book.by_id(ROW_ID)
356        await book.cover.append_file_from_url("https://example.com/image.png")
357        book.update()
358        ```
359
360        Args:
361            url (str): The URL of the file.
362            name (str | None): Optional human-readable name of the file, as it
363                should be displayed in the Baserow frontend.
364            client (Client | None): Specifies which API client should be used.
365                If none is provided, the `GlobalClient` will be used. Note that
366                this method does not automatically use the client associated
367                with the table.
368            register_pending_change (bool): Internal option for the
369                `FileField.from_file()` and `FileField.from_url()` methods. When
370                set to false, tracking of pending changes for this call is
371                disabled. This is only useful in situations where
372                `Table.update()` will be called anyway, such as when the method
373                is invoked within the initialization of a model.
374        """
375        if client is None:
376            client = GlobalClient()
377        new_file = await client.upload_file_via_url(url)
378        if name is not None:
379            new_file.original_name = name
380        self.root.append(new_file)
381        if register_pending_change:
382            self.register_pending_change(f"file from url '{url}' added")
383
384    def clear(self):
385        """
386        Removes all files from field. After that, `Table.update()` must be called
387        to apply the changes.
388
389        ```python
390        book = Book.by_id(ROW_ID)
391        await book.cover.clear()
392        book.update()
393        ```
394        """
395        self.root.clear()
396        self.register_pending_change("remove all files in field")
397
398
399SelectEnum = TypeVar("SelectEnum", bound=enum.Enum)
400"""
401Instances of a SelectEntry have to be bound to a enum which contain the possible
402values of the select entry.
403"""
404
405
406class SelectEntry(BaseModel, Generic[SelectEnum]):
407    """A entry in a single or multiple select field."""
408    entry_id: Optional[int] = Field(default=None, alias="id")
409    value: Optional[SelectEnum] = None
410    color: Optional[str] = None
411
412    model_config = ConfigDict(populate_by_name=True)
413
414    @model_validator(mode="after")
415    def id_or_value_must_be_set(self: "SelectEntry") -> "SelectEntry":
416        if self.entry_id is None and self.value is None:
417            raise ValueError(
418                "At least one of the entry_id and value fields must be set"
419            )
420        return self
421
422    @model_serializer
423    def serialize(self) -> Union[int, str]:
424        """
425        Serializes the field into the data structure required by the Baserow
426        API. If an entry has both an id and a value set, the id is used.
427        Otherwise the set field is used.
428
429        From the Baserow API documentation: Accepts an integer or a text value
430        representing the chosen select option id or option value. A null value
431        means none is selected. In case of a text value, the first matching
432        option is selected. 
433        """
434        if self.entry_id is not None:
435            return self.entry_id
436        if self.value is not None:
437            return self.value.value
438        raise ValueError("both fields id and value are unset for this entry")
439
440    @classmethod
441    def _get_all_possible_values(cls) -> list[str]:
442        metadata = cls.__pydantic_generic_metadata__
443        if "args" not in metadata:
444            raise PydanticGenericMetadataError.args_missing(
445                cls.__class__.__name__,
446                "select entry enum",
447            )
448        if len(metadata["args"]) < 1:
449            raise PydanticGenericMetadataError.args_empty(
450                cls.__class__.__name__,
451                "select entry enum",
452            )
453        select_enum = metadata["args"][0]
454        return [item.value for item in select_enum]
455
456    @classmethod
457    def _options_config(cls) -> list[SelectEntryConfig]:
458        rsl: list[SelectEntryConfig] = []
459        color_sequence = ColorSequence()
460        i = 0
461        for value in cls._get_all_possible_values():
462            rsl.append(SelectEntryConfig(
463                id=i,
464                value=value,
465                color=color_sequence.get_color(),
466            ))
467            i += 1
468        return rsl
469
470
471class SingleSelectField(SelectEntry[SelectEnum], BaserowField):
472    """Single select field in a table."""
473    @classmethod
474    def default_config(cls) -> FieldConfigType:
475        options = super(SingleSelectField, cls)._options_config()
476        return SingleSelectFieldConfig(select_options=options)
477
478    @classmethod
479    def read_only(cls) -> bool:
480        return False
481
482    @classmethod
483    def from_enum(cls, select_enum: SelectEnum):
484        """
485        This function can be used to directly obtain the correct instance of the
486        field abstraction from an enum. Primarily, this function is a quality of
487        life feature for directly setting a field value in a model
488        initialization. This replaces the somewhat unergonomic and unintuitive
489        syntax which would be used otherwise.
490
491        ```python
492        class Genre(str, enum.Enum):
493            FICTION = "Fiction"
494            EDUCATION = "Education"
495
496        class Book(Table):
497            [...]
498            genre: Optional[SingleSelectField[Genre]] = Field(default=None)
499
500        # Can use this...
501        await Book(
502            genre=SingleSelectField.from_enum(Genre.FICTION),
503        ).create()
504
505        # ...instead of
506        await Book(
507            genre=SingleSelectField[Genre](value=Genre.FICTION)
508        ).create()
509        ```
510
511        Args:
512            select_enum (SelectEnum): Enum to which the field should be set.add 
513        """
514        return SingleSelectField[type(select_enum)](value=select_enum)
515
516    def set(self, instance: SelectEnum):
517        """
518        Set the value of the field. Please note that this method does not update
519        the record on Baserow. You have to call `Table.update()` afterwards.
520
521        Args:
522            instance (SelectEnum): The enum which should be set in this field.
523        """
524        self.entry_id = None
525        self.value = instance
526        self.color = None
527        self.register_pending_change(f"set SingleSelect to '{instance.value}'")
528
529
530class MultipleSelectField(BaserowField, RootModel[list[SelectEntry]], Generic[SelectEnum]):
531    """Multiple select field in a table."""
532    root: list[SelectEntry[SelectEnum]]
533
534    @classmethod
535    def read_only(cls) -> bool:
536        return False
537
538    @classmethod
539    def from_enums(cls, *select_enums: SelectEnum):
540        """
541        This function can be used to directly obtain the correct instance of the
542        field abstraction from one or multiple enum(s). Primarily, this function
543        is a quality of life feature for directly setting a field value in a
544        model initialization. This replaces the somewhat unergonomic and
545        unintuitive syntax which would be used otherwise.
546
547        ```python
548        class Keyword(str, enum.Enum):
549            ADVENTURE = "Adventure"
550            FICTION = "Fiction"
551            TECH = "Text"
552
553        class Book(Table):
554            [...]
555            keywords: Optional[MultipleSelectField[Keyword]] = Field(default=None)
556
557        await Book(
558            keywords=MultipleSelectField.from_enums(Keyword.ADVENTURE, Keyword.FICTION)
559        ).create()
560        ```
561
562        Args:
563            *select_enum (SelectEnum | list[SelectEnum]): Enum(s) value(s) to be part
564                of the select field.
565        """
566        if not select_enums:
567            raise ValueError("At least one enum must be provided")
568        select_enum_type = type(select_enums[0])
569        enums: list[SelectEntry[SelectEnum]] = []
570        for enum in select_enums:
571            enums.append(SelectEntry[type(enum)](value=enum))
572        return MultipleSelectField[select_enum_type](root=enums)
573
574    def append(self, *select_enums: SelectEnum):
575        """
576        Append one or multiple enum(s) to the field. Please note that this
577        method does not update the record on Baserow. You have to call
578        `Table.update()` afterwards.
579
580        Args:
581            *select_enum (SelectEnum | list[SelectEnum]): Enum(s) value(s) to be
582                added the the select field.
583        """
584        names: list[str] = []
585        for enum in select_enums:
586            self.root.append(SelectEntry(value=enum))
587            names.append(enum.name)
588        self.register_pending_change(
589            f"append enum(s) {', '.join(names)} to multiple select field",
590        )
591
592    def clear(self):
593        """
594        Remove all enums currently set for this field. Please note that this
595        method does not update the record on Baserow. You have to call
596        `Table.update()` afterwards.
597        """
598        self.root.clear()
599        self.register_pending_change(
600            f"removed all enum(s) from multiple select field",
601        )
602
603    def remove(self, *select_enums: SelectEnum):
604        """
605        Remove one or multiple enum(s) from the field. Please note that this
606        method does not update the record on Baserow. You have to call
607        `Table.update()` afterwards.
608
609        Args:
610            *select_enum (SelectEnum | list[SelectEnum]): Enum(s) value(s) to be
611                removed the the select field.
612        """
613        to_be_removed = [
614            entry for entry in self.root if entry. value in select_enums]
615        self.root = [
616            entry for entry in self.root if entry not in to_be_removed
617        ]
618        names = [
619            entry.value.name for entry in to_be_removed if entry.value is not None]
620        self.register_pending_change(
621            f"removed enum(s) {', '.join(names)} from multiple select field",
622        )
623
624    @classmethod
625    def default_config(cls) -> FieldConfigType:
626        metadata = cls.__pydantic_generic_metadata__
627        if "args" not in metadata:
628            raise PydanticGenericMetadataError.args_missing(
629                cls.__class__.__name__,
630                "select entry enum",
631            )
632        if len(metadata["args"]) < 1:
633            raise PydanticGenericMetadataError.args_empty(
634                cls.__class__.__name__,
635                "select entry enum",
636            )
637        select_enum = metadata["args"][0]
638        rsl: list[SelectEntryConfig] = []
639        color_sequence = ColorSequence()
640        i = 0
641        for item in select_enum:
642            rsl.append(SelectEntryConfig(
643                id=i,
644                value=item.value,
645                color=color_sequence.get_color(),
646            ))
647            i += 1
648        return MultipleSelectFieldConfig(select_options=rsl)