Skip to content

Models

Info

Models serve as the primary mechanism for metadata harvesting. By extending pydantic, defining data models and validating input data becomes effortless.

Base

Bases: pydantic.BaseModel

Base model provides efficient model persistence and data validation capabilities. The persistence mechanism relies on sqlite and pickle, allowing the entire model to be stored as a snapshot

Usage:

class MyModel(BaseModel):
    title: str

# store a snapshot of the model
stored_model = MyModel(title="Model")
stored_model.save()

# we should be able to retrieve the same model
assert MyModel.all() == [stored_model] # True
Source code in nucleus/sdk/harvest/models.py 
 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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class Base(pydantic.BaseModel, metaclass=_Manager):
    """Base model provides efficient model persistence and data validation capabilities.
    The persistence mechanism relies on sqlite and pickle, allowing the entire model to be stored as a snapshot

    Usage:

        class MyModel(BaseModel):
            title: str

        # store a snapshot of the model
        stored_model = MyModel(title="Model")
        stored_model.save()

        # we should be able to retrieve the same model
        assert MyModel.all() == [stored_model] # True
    """

    _alias: str
    _conn: Connection

    class Config:
        # Frozen model behavior
        # ref: https://docs.pydantic.dev/usage/model_config/
        frozen = True
        smart_union = True
        use_enum_values = True
        arbitrary_types_allowed = True
        anystr_strip_whitespace = True

    def __init__(self, *args: Any, **kwargs: Any):
        try:
            super().__init__(*args, **kwargs)
        except ValidationError as e:
            raise ModelValidationError(f'raised exception during model initialization: {str(e)}')

        sqlite3.register_converter(self._alias, pickle.loads)
        sqlite3.register_adapter(type(self), pickle.dumps)

    @classmethod
    @decorators.proxy_exception(
        expected=sqlite3.ProgrammingError,
        target=ModelManagerError,
    )
    def get(cls) -> Base:
        """Exec query and fetch first entry from local database.

        :return: First registered snapshot
        :raises ModelManagerError: If there is an error fetching entry
        """

        response = cls._conn.execute(FETCH % cls._alias)
        row = response.fetchone()
        return row[0]

    @classmethod
    @decorators.proxy_exception(
        expected=sqlite3.ProgrammingError,
        target=ModelManagerError,
    )
    def all(cls) -> Iterator[Base]:
        """Exec query and fetch a list of data from local database.

        :return: all registered snapshots
        :raises ModelManagerError: If there is an error fetching entries
        """
        response = cls._conn.execute(FETCH % cls._alias)
        rows = response.fetchall()
        return map(lambda r: r[0], rows)

    @decorators.proxy_exception(
        expected=sqlite3.ProgrammingError,
        target=ModelManagerError,
    )
    def save(self) -> bool:
        """Exec insertion query into local database

        :return: True if successful else False
        :raises ModelManagerError: If there is an error saving entry
        """

        # https://docs.python.org/3/library/sqlite3.html#sqlite3.Cursor.lastrowid
        cursor = self._conn.execute(INSERT % self._alias, (self,))
        return cursor.rowcount > 0

all() classmethod

Exec query and fetch a list of data from local database.

Returns:

Type Description
Iterator[Base]

all registered snapshots

Raises:

Type Description
ModelManagerError

If there is an error fetching entries
Source code in nucleus/sdk/harvest/models.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@classmethod
@decorators.proxy_exception(
    expected=sqlite3.ProgrammingError,
    target=ModelManagerError,
)
def all(cls) -> Iterator[Base]:
    """Exec query and fetch a list of data from local database.

    :return: all registered snapshots
    :raises ModelManagerError: If there is an error fetching entries
    """
    response = cls._conn.execute(FETCH % cls._alias)
    rows = response.fetchall()
    return map(lambda r: r[0], rows)

get() classmethod

Exec query and fetch first entry from local database.

Returns:

Type Description
Base

First registered snapshot

Raises:

Type Description
ModelManagerError

If there is an error fetching entry
Source code in nucleus/sdk/harvest/models.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@classmethod
@decorators.proxy_exception(
    expected=sqlite3.ProgrammingError,
    target=ModelManagerError,
)
def get(cls) -> Base:
    """Exec query and fetch first entry from local database.

    :return: First registered snapshot
    :raises ModelManagerError: If there is an error fetching entry
    """

    response = cls._conn.execute(FETCH % cls._alias)
    row = response.fetchone()
    return row[0]

save()

Exec insertion query into local database

Returns:

Type Description
bool

True if successful else False

Raises:

Type Description
ModelManagerError

If there is an error saving entry
Source code in nucleus/sdk/harvest/models.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@decorators.proxy_exception(
    expected=sqlite3.ProgrammingError,
    target=ModelManagerError,
)
def save(self) -> bool:
    """Exec insertion query into local database

    :return: True if successful else False
    :raises ModelManagerError: If there is an error saving entry
    """

    # https://docs.python.org/3/library/sqlite3.html#sqlite3.Cursor.lastrowid
    cursor = self._conn.execute(INSERT % self._alias, (self,))
    return cursor.rowcount > 0

Media

Bases: Base, Generic[T]

Generic media model to create media subtypes. Each subtype represents a specific media type and provides a generic specification of the sources from which it can be collected.

Usage:

class Video(Media[Path]):
    # represents a video file type .
    ...

class Image(Media[URL]):
    # represents an image url type.
    ...
Source code in nucleus/sdk/harvest/models.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class Media(Base, Generic[T]):
    """Generic media model to create media subtypes.
    Each subtype represents a specific media type and provides a generic specification
    of the sources from which it can be collected.

    Usage:

        class Video(Media[Path]):
            # represents a video file type .
            ...

        class Image(Media[URL]):
            # represents an image url type.
            ...
    """

    path: T

Model

Bases: Base

Model class specifies by default the attributes needed for the metadata model and allows its extension to create metadata sub-models with custom attributes.

Usage:

class Nucleus(Model):
    title: str # default property
    description: str # default property
    address: str # my custom property
Source code in nucleus/sdk/harvest/models.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class Model(Base):
    """Model class specifies by default the attributes needed for the metadata model
    and allows its extension to create metadata sub-models with custom attributes.

    Usage:

        class Nucleus(Model):
            title: str # default property
            description: str # default property
            address: str # my custom property
    """

    title: str  # the name of the resource
    description: str  # the description of the resource