Classes

class Schema__Table__Files (**kwargs)

A mixin class to strictly assign keyword arguments to pre-defined instance attributes during initialization.

This base class provides an init method that assigns values from keyword arguments to instance attributes. If an attribute with the same name as a key from the kwargs is defined in the class, it will be set to the value from kwargs. If the key does not match any predefined attribute names, an exception is raised.

This behavior enforces strict control over the attributes of instances, ensuring that only predefined attributes can be set at the time of instantiation and avoids silent attribute creation which can lead to bugs in the code.

Usage

class MyConfigurableClass(Kwargs_To_Self): attribute1 = 'default_value' attribute2 = True attribute3 : str attribute4 : list attribute4 : int = 42

# Other methods can be added here

Correctly override default values by passing keyword arguments

instance = MyConfigurableClass(attribute1='new_value', attribute2=False)

This will raise an exception as 'attribute3' is not predefined

instance = MyConfigurableClass(attribute3='invalid_attribute')

this will also assign the default value to any variable that has a type defined. In the example above the default values (mapped by default__kwargs and locals) will be: attribute1 = 'default_value' attribute2 = True attribute3 = '' # default value of str attribute4 = [] # default value of list attribute4 = 42 # defined value in the class

Note

It is important that all attributes which may be set at instantiation are predefined in the class. Failure to do so will result in an exception being raised.

Methods

init(**kwargs): The initializer that handles the assignment of keyword arguments to instance attributes. It enforces strict attribute assignment rules, only allowing attributes that are already defined in the class to be set.

Initialize an instance of the derived class, strictly assigning provided keyword arguments to corresponding instance attributes.

Parameters

**kwargs: Variable length keyword arguments.

Raises

Exception

If a key from kwargs does not correspond to any attribute pre-defined in the class, an exception is raised to prevent setting an undefined attribute.

Expand source code

class Schema__Table__Files(Kwargs_To_Self):
    path     : str
    contents : bytes
    metadata : bytes
    timestamp: int

Ancestors

• Kwargs_To_Self

Class variables

var contents : bytes

var metadata : bytes

var path : str

var timestamp : int

Inherited members

• Kwargs_To_Self:
  • update_from_kwargs

class Sqlite__Table__Files (**kwargs)

A mixin class to strictly assign keyword arguments to pre-defined instance attributes during initialization.

This base class provides an init method that assigns values from keyword arguments to instance attributes. If an attribute with the same name as a key from the kwargs is defined in the class, it will be set to the value from kwargs. If the key does not match any predefined attribute names, an exception is raised.

This behavior enforces strict control over the attributes of instances, ensuring that only predefined attributes can be set at the time of instantiation and avoids silent attribute creation which can lead to bugs in the code.

Usage

class MyConfigurableClass(Kwargs_To_Self): attribute1 = 'default_value' attribute2 = True attribute3 : str attribute4 : list attribute4 : int = 42

# Other methods can be added here

Correctly override default values by passing keyword arguments

instance = MyConfigurableClass(attribute1='new_value', attribute2=False)

This will raise an exception as 'attribute3' is not predefined

instance = MyConfigurableClass(attribute3='invalid_attribute')

this will also assign the default value to any variable that has a type defined. In the example above the default values (mapped by default__kwargs and locals) will be: attribute1 = 'default_value' attribute2 = True attribute3 = '' # default value of str attribute4 = [] # default value of list attribute4 = 42 # defined value in the class

Note

It is important that all attributes which may be set at instantiation are predefined in the class. Failure to do so will result in an exception being raised.

Methods

init(**kwargs): The initializer that handles the assignment of keyword arguments to instance attributes. It enforces strict attribute assignment rules, only allowing attributes that are already defined in the class to be set.

Initialize an instance of the derived class, strictly assigning provided keyword arguments to corresponding instance attributes.

Parameters

**kwargs: Variable length keyword arguments.

Raises

Exception

If a key from kwargs does not correspond to any attribute pre-defined in the class, an exception is raised to prevent setting an undefined attribute.

Expand source code

class Sqlite__Table__Files(Sqlite__Table):
    auto_pickle_blob    : bool = True
    set_timestamp       : bool = True

    def __init__(self, **kwargs):
        self.table_name = SQLITE__TABLE_NAME__FILES
        self.row_schema  = Schema__Table__Files
        super().__init__(**kwargs)

    def add_file(self, path, contents=None, metadata= None):
        if self.contains(path=path):                                                    # don't allow multiple entries for the same file path (until we add versioning support)
            return status_warning(f"File not added, since file with path '{path}' already exists in the database")
        if metadata is None:
            metadata = {}
        metadata.update(self.create_contents_metadata(contents))
        row_data    = self.create_node_data(path, contents, metadata)
        new_row_obj = self.add_row_and_commit(**row_data)
        return status_ok(message='file added', data= new_row_obj)

    def create_contents_metadata(self, contents):
        file_size       = len(contents)
        file_is_binary = type(contents) is bytes
        if file_is_binary:
            file_hash  = bytes_sha256(contents)
        else:
            file_hash = str_sha256(str(contents))
        return dict(file_contents=dict(hash      = file_hash      ,
                                       is_binary = file_is_binary ,
                                       size      = file_size      ))

    def delete_file(self, path):
        if self.not_contains(path=path):                                                    # don't allow multiple entries for the same file path (until we add versioning support)
            return status_warning(f"File not deleted, since file with path '{path}' did not exist in the database")

        self.rows_delete_where(path=path)
        return status_ok(message='file deleted')

    def create_node_data(self, path, contents=None, metadata= None):
        node_data = {'path'    : path     ,
                     'contents': contents ,
                     'metadata': metadata }
        if self.set_timestamp:
            node_data['timestamp'] = timestamp_utc_now()
        return node_data

    def field_names_without_content(self):                      # todo: refactor to get this directly from the schema
        return ['id', 'path', 'metadata', 'timestamp']          #       and so that these values are not hard-coded here

    def file(self, path, include_contents=True):
        if include_contents:
            fields = ['*']
        else:
            fields = self.field_names_without_content()
        return self.row(where=dict(path=path), fields = fields)

    def file_without_contents(self, path):
        return self.file(path, include_contents=False)

    def file_exists(self, path):
        return self.contains(path=path)

    def files(self, include_contents=False):
        if include_contents:
            return self.rows()
        fields_names = self.field_names_without_content()
        return self.rows(fields_names)

    def setup(self):
        if self.exists() is False:
            self.create()
            self.index_create('path')
        return self

Ancestors

• Sqlite__Table
• Kwargs_To_Self

Class variables

var auto_pickle_blob : bool

var set_timestamp : bool

Methods

def add_file(self, path, contents=None, metadata=None)

Expand source code

def add_file(self, path, contents=None, metadata= None):
    if self.contains(path=path):                                                    # don't allow multiple entries for the same file path (until we add versioning support)
        return status_warning(f"File not added, since file with path '{path}' already exists in the database")
    if metadata is None:
        metadata = {}
    metadata.update(self.create_contents_metadata(contents))
    row_data    = self.create_node_data(path, contents, metadata)
    new_row_obj = self.add_row_and_commit(**row_data)
    return status_ok(message='file added', data= new_row_obj)

def create_contents_metadata(self, contents)

Expand source code

def create_contents_metadata(self, contents):
    file_size       = len(contents)
    file_is_binary = type(contents) is bytes
    if file_is_binary:
        file_hash  = bytes_sha256(contents)
    else:
        file_hash = str_sha256(str(contents))
    return dict(file_contents=dict(hash      = file_hash      ,
                                   is_binary = file_is_binary ,
                                   size      = file_size      ))

def create_node_data(self, path, contents=None, metadata=None)

Expand source code

def create_node_data(self, path, contents=None, metadata= None):
    node_data = {'path'    : path     ,
                 'contents': contents ,
                 'metadata': metadata }
    if self.set_timestamp:
        node_data['timestamp'] = timestamp_utc_now()
    return node_data

def delete_file(self, path)

Expand source code

def delete_file(self, path):
    if self.not_contains(path=path):                                                    # don't allow multiple entries for the same file path (until we add versioning support)
        return status_warning(f"File not deleted, since file with path '{path}' did not exist in the database")

    self.rows_delete_where(path=path)
    return status_ok(message='file deleted')

def field_names_without_content(self)

Expand source code

def field_names_without_content(self):                      # todo: refactor to get this directly from the schema
    return ['id', 'path', 'metadata', 'timestamp']          #       and so that these values are not hard-coded here

def file(self, path, include_contents=True)

Expand source code

def file(self, path, include_contents=True):
    if include_contents:
        fields = ['*']
    else:
        fields = self.field_names_without_content()
    return self.row(where=dict(path=path), fields = fields)

def file_exists(self, path)

Expand source code

def file_exists(self, path):
    return self.contains(path=path)

def file_without_contents(self, path)

Expand source code

def file_without_contents(self, path):
    return self.file(path, include_contents=False)

def files(self, include_contents=False)

Expand source code

def files(self, include_contents=False):
    if include_contents:
        return self.rows()
    fields_names = self.field_names_without_content()
    return self.rows(fields_names)

def setup(self)

Expand source code

def setup(self):
    if self.exists() is False:
        self.create()
        self.index_create('path')
    return self

Inherited members

• Sqlite__Table:
  • update_from_kwargs