Module osbot_utils.helpers.sqlite.Sqlite__Table__Create

Expand source code
import inspect
from typing import List

from osbot_utils.base_classes.Kwargs_To_Self                import Kwargs_To_Self
from osbot_utils.decorators.lists.filter_list               import filter_list
from osbot_utils.helpers.sqlite.Sqlite__Field               import Sqlite__Field
from osbot_utils.helpers.sqlite.Sqlite__Table               import Sqlite__Table, DEFAULT_FIELD_NAME__ID
from osbot_utils.helpers.sqlite.models.Sqlite__Field__Type  import Sqlite__Field__Type

class Sqlite__Table__Create(Kwargs_To_Self):
    fields  : List[Sqlite__Field]
    table   : Sqlite__Table

    def __init__(self, table_name):
        super().__init__()
        self.table.table_name = table_name
        self.set_default_fields()

    def add_field(self, field_data: dict):
        sqlite_field = Sqlite__Field.from_json(field_data)
        if sqlite_field:
            self.fields.append(sqlite_field)
            return True
        return False

    def add_fields(self, fields_data:List[dict]):
        results = []
        if fields_data:
            for field_data in fields_data:
                results.append(self.add_field(field_data))
        return results

    def add_fields_from_class(self, table_class):
        if inspect.isclass(table_class):
            for field_name, field_type in table_class.__annotations__.items():
                self.add_field_with_type(field_name, field_type)
        return self

    def add_field_with_type(self, field_name, field_type):
        if inspect.isclass(field_type):
            field_type = Sqlite__Field__Type.type_map().get(field_type)

        return self.add_field(dict(name=field_name, type=field_type))

    def add_field__text(self, field_name):
        return self.add_field_with_type(field_name=field_name, field_type=str)

    def add_fields__text(self, *fields_name):
        for field_name in fields_name:
            self.add_field__text(field_name=field_name)
        return self

    def create_table(self):
        sql_query = self.sql_for__create_table()
        if self.table.not_exists():
            self.table.cursor().execute(sql_query)
            return self.table.exists()
        return False

    def create_table__from_row_schema(self, row_schema):  # todo add check if there is an index field (which is now supported since it clashes with the one that is added by default)
        self.add_fields_from_class(row_schema)
        self.table.row_schema = row_schema
        return self.create_table()

    @filter_list
    def fields_json(self):
        return [field.json() for field in self.fields]

    def fields__by_name_type(self):
        return { item.get('name'): item.get('type') for item in self.fields_json() }

    def fields_reset(self):
        self.fields = []
        return self

    def database(self):
        return self.table.database

    def set_default_fields(self):
        self.add_field(dict(name=DEFAULT_FIELD_NAME__ID, type="INTEGER", pk=True))        # by default every table has an id field
        return self

    def sql_for__create_table(self):
        field_definitions = [field.text_for_create_table() for field in self.fields]
        primary_keys = [field.name for field in self.fields if field.pk]
        foreign_keys_constraints = [field.text_for_create_table() for field in self.fields if field.is_foreign_key]

        # Handling composite primary keys if necessary
        if len(primary_keys) > 1:
            pk_constraint = f"PRIMARY KEY ({', '.join(primary_keys)})"
            field_definitions.append(pk_constraint)

        # Adding foreign key constraints separately if there are any
        if foreign_keys_constraints:
            field_definitions.extend(foreign_keys_constraints)

        table_definition = f"CREATE TABLE {self.table.table_name} ({', '.join(field_definitions)});"
        return table_definition

Classes

class Sqlite__Table__Create (table_name)

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__Create(Kwargs_To_Self):
    fields  : List[Sqlite__Field]
    table   : Sqlite__Table

    def __init__(self, table_name):
        super().__init__()
        self.table.table_name = table_name
        self.set_default_fields()

    def add_field(self, field_data: dict):
        sqlite_field = Sqlite__Field.from_json(field_data)
        if sqlite_field:
            self.fields.append(sqlite_field)
            return True
        return False

    def add_fields(self, fields_data:List[dict]):
        results = []
        if fields_data:
            for field_data in fields_data:
                results.append(self.add_field(field_data))
        return results

    def add_fields_from_class(self, table_class):
        if inspect.isclass(table_class):
            for field_name, field_type in table_class.__annotations__.items():
                self.add_field_with_type(field_name, field_type)
        return self

    def add_field_with_type(self, field_name, field_type):
        if inspect.isclass(field_type):
            field_type = Sqlite__Field__Type.type_map().get(field_type)

        return self.add_field(dict(name=field_name, type=field_type))

    def add_field__text(self, field_name):
        return self.add_field_with_type(field_name=field_name, field_type=str)

    def add_fields__text(self, *fields_name):
        for field_name in fields_name:
            self.add_field__text(field_name=field_name)
        return self

    def create_table(self):
        sql_query = self.sql_for__create_table()
        if self.table.not_exists():
            self.table.cursor().execute(sql_query)
            return self.table.exists()
        return False

    def create_table__from_row_schema(self, row_schema):  # todo add check if there is an index field (which is now supported since it clashes with the one that is added by default)
        self.add_fields_from_class(row_schema)
        self.table.row_schema = row_schema
        return self.create_table()

    @filter_list
    def fields_json(self):
        return [field.json() for field in self.fields]

    def fields__by_name_type(self):
        return { item.get('name'): item.get('type') for item in self.fields_json() }

    def fields_reset(self):
        self.fields = []
        return self

    def database(self):
        return self.table.database

    def set_default_fields(self):
        self.add_field(dict(name=DEFAULT_FIELD_NAME__ID, type="INTEGER", pk=True))        # by default every table has an id field
        return self

    def sql_for__create_table(self):
        field_definitions = [field.text_for_create_table() for field in self.fields]
        primary_keys = [field.name for field in self.fields if field.pk]
        foreign_keys_constraints = [field.text_for_create_table() for field in self.fields if field.is_foreign_key]

        # Handling composite primary keys if necessary
        if len(primary_keys) > 1:
            pk_constraint = f"PRIMARY KEY ({', '.join(primary_keys)})"
            field_definitions.append(pk_constraint)

        # Adding foreign key constraints separately if there are any
        if foreign_keys_constraints:
            field_definitions.extend(foreign_keys_constraints)

        table_definition = f"CREATE TABLE {self.table.table_name} ({', '.join(field_definitions)});"
        return table_definition

Ancestors

Class variables

var fields : List[Sqlite__Field]
var tableSqlite__Table

Methods

def add_field(self, field_data: dict)
Expand source code
def add_field(self, field_data: dict):
    sqlite_field = Sqlite__Field.from_json(field_data)
    if sqlite_field:
        self.fields.append(sqlite_field)
        return True
    return False
def add_field__text(self, field_name)
Expand source code
def add_field__text(self, field_name):
    return self.add_field_with_type(field_name=field_name, field_type=str)
def add_field_with_type(self, field_name, field_type)
Expand source code
def add_field_with_type(self, field_name, field_type):
    if inspect.isclass(field_type):
        field_type = Sqlite__Field__Type.type_map().get(field_type)

    return self.add_field(dict(name=field_name, type=field_type))
def add_fields(self, fields_data: List[dict])
Expand source code
def add_fields(self, fields_data:List[dict]):
    results = []
    if fields_data:
        for field_data in fields_data:
            results.append(self.add_field(field_data))
    return results
def add_fields__text(self, *fields_name)
Expand source code
def add_fields__text(self, *fields_name):
    for field_name in fields_name:
        self.add_field__text(field_name=field_name)
    return self
def add_fields_from_class(self, table_class)
Expand source code
def add_fields_from_class(self, table_class):
    if inspect.isclass(table_class):
        for field_name, field_type in table_class.__annotations__.items():
            self.add_field_with_type(field_name, field_type)
    return self
def create_table(self)
Expand source code
def create_table(self):
    sql_query = self.sql_for__create_table()
    if self.table.not_exists():
        self.table.cursor().execute(sql_query)
        return self.table.exists()
    return False
def create_table__from_row_schema(self, row_schema)
Expand source code
def create_table__from_row_schema(self, row_schema):  # todo add check if there is an index field (which is now supported since it clashes with the one that is added by default)
    self.add_fields_from_class(row_schema)
    self.table.row_schema = row_schema
    return self.create_table()
def database(self)
Expand source code
def database(self):
    return self.table.database
def fields__by_name_type(self)
Expand source code
def fields__by_name_type(self):
    return { item.get('name'): item.get('type') for item in self.fields_json() }
def fields_json(self)
Expand source code
@filter_list
def fields_json(self):
    return [field.json() for field in self.fields]
def fields_reset(self)
Expand source code
def fields_reset(self):
    self.fields = []
    return self
def set_default_fields(self)
Expand source code
def set_default_fields(self):
    self.add_field(dict(name=DEFAULT_FIELD_NAME__ID, type="INTEGER", pk=True))        # by default every table has an id field
    return self
def sql_for__create_table(self)
Expand source code
def sql_for__create_table(self):
    field_definitions = [field.text_for_create_table() for field in self.fields]
    primary_keys = [field.name for field in self.fields if field.pk]
    foreign_keys_constraints = [field.text_for_create_table() for field in self.fields if field.is_foreign_key]

    # Handling composite primary keys if necessary
    if len(primary_keys) > 1:
        pk_constraint = f"PRIMARY KEY ({', '.join(primary_keys)})"
        field_definitions.append(pk_constraint)

    # Adding foreign key constraints separately if there are any
    if foreign_keys_constraints:
        field_definitions.extend(foreign_keys_constraints)

    table_definition = f"CREATE TABLE {self.table.table_name} ({', '.join(field_definitions)});"
    return table_definition

Inherited members