Module osbot_utils.helpers.sqlite.sample_data.Sqlite__Sample_Data__Chinook

Expand source code
from osbot_utils.base_classes.Kwargs_To_Self import Kwargs_To_Self
from osbot_utils.decorators.methods.cache_on_self import cache_on_self
from osbot_utils.helpers.sqlite.Sqlite__Database import Sqlite__Database
from osbot_utils.helpers.sqlite.Sqlite__Table import Sqlite__Table
from osbot_utils.helpers.sqlite.Sqlite__Table__Create import Sqlite__Table__Create
from osbot_utils.helpers.sqlite.domains.Sqlite__DB__Json import Sqlite__DB__Json
from osbot_utils.testing.Duration import duration, Duration
from osbot_utils.utils.Dev import pprint
from osbot_utils.utils.Files import current_temp_folder, path_combine, folder_create, file_exists, file_not_exists, \
    file_contents
from osbot_utils.utils.Http import GET, GET_to_file
from osbot_utils.utils.Json import json_loads, json_from_file, json_dump, json_file_create, json_file_load
from osbot_utils.utils.Objects import obj_info

URL__CHINOOK_JSON                = 'https://github.com/lerocha/chinook-database/releases/download/v1.4.5/ChinookData.json'
FILE_NAME__CHINOOK_DATA_JSON     = 'ChinookData.json'
FILE_NAME__TABLES_SCHEMAS_FIELDS = 'chinook__tables_schemas_fields.json'
FOLDER_NAME__CHINOOK_DATA        = 'chinook_data'
FOLDER_NAME__SQLITE_DATA_SETS    = '_sqlite_data_sets'
PATH__DB__TESTS                  = '/tmp/db-tests'
PATH__DB__CHINOOK                = '/tmp/db-tests/test-chinook.db'
TABLE_NAME__CHINOOK              = 'chinook_data'

class Sqlite__Sample_Data__Chinook(Kwargs_To_Self):
    url_chinook_database_json : str                 = URL__CHINOOK_JSON
    path_local_db             : str                 = PATH__DB__CHINOOK
    table_name                : str                 = TABLE_NAME__CHINOOK
    json_db                   : Sqlite__DB__Json



    def chinook_data_as_json(self):
        path_chinook_data_as_json = self.path_chinook_data_as_json()
        if file_not_exists(path_chinook_data_as_json):
            GET_to_file(self.url_chinook_database_json, path_chinook_data_as_json)
        return json_from_file(path_chinook_data_as_json)

    # def create_table_from_data(self):
    #     chinook_data = self.chinook_data_as_json()
    #     table_creator = Sqlite__Table__Create(table_name=self.table_name)
    #     table         = table_creator.table
    #     table_creator.add_fields__text("name", "value").create_table()
    #
    #     cursor = table.cursor()
    #     assert len(chinook_data) == 11
    #     for key, items in chinook_data.items():
    #         name = key
    #         value = json_dump(items)
    #         table.row_add(dict(name=name, value=value))
    #
    #     cursor.commit()
    #     return table

    def create_tables(self):
        self.create_tables_from_schema()
        chinook_data = self.chinook_data_as_json()
        for table_name, rows in chinook_data.items():
            table = self.json_db.database.table(table_name)
            table.rows_add(rows)



    def create_tables_from_schema(self):
        schemas  = self.tables_schemas_fields_from_data()
        for table_name, table_schema in schemas.items():
            self.json_db.create_table_from_schema(table_name, table_schema)
        return self

    def database(self):                         # todo: need to refactor this code based from the creation of the database to the use of it
        return self.json_db.database

    @cache_on_self
    def load_db_from_disk(self):
        db_chinook = Sqlite__Database(db_path=PATH__DB__CHINOOK, auto_schema_row=True)          # set the auto_schema_row so that we have a row_schema defined for all tables
        return db_chinook
        # db_chinook.connect()
        # return db_chinook.table(self.table_name)

    def path_chinook_data_as_json(self):
        return path_combine(self.path_chinook_data(), FILE_NAME__CHINOOK_DATA_JSON)

    def path_chinook_data(self):
        path_chinook_data = path_combine(self.path_sqlite_sample_data_sets(), FOLDER_NAME__CHINOOK_DATA)
        return folder_create(path_chinook_data)

    def path_sqlite_sample_data_sets(self):                     # todo: refactor to sqlite_sample_data_sets helper class
        path_data_sets = path_combine(current_temp_folder(), FOLDER_NAME__SQLITE_DATA_SETS)
        return folder_create(path_data_sets)

    def path_tables_schemas_fields(self):
        return path_combine(self.path_chinook_data(), FILE_NAME__TABLES_SCHEMAS_FIELDS)

    def table__genre(self):
        return self.load_db_from_disk().table('Genre')          # todo: refactor to sqlite_sample_data_sets helper class

    def table__track(self):
        return self.load_db_from_disk().table('Track')          # todo: refactor to sqlite_sample_data_sets helper class

    # def tables(self):
    #     return self.load_db_from_disk().tables()

    def tables_schemas_fields_from_data(self):
        path_tables_schemas_fields = self.path_tables_schemas_fields()
        if file_not_exists(path_tables_schemas_fields):
            tables_schemas = {}
            for name, data  in self.chinook_data_as_json().items():
                table_schema = self.json_db.get_schema_from_json_data(data)
                tables_schemas[name] = table_schema
            json_file_create(tables_schemas, path=path_tables_schemas_fields)

        return json_file_load(path_tables_schemas_fields)

    def save(self, path=PATH__DB__CHINOOK):
        database = self.database()
        database.save_to(path)
        return True

Classes

class Sqlite__Sample_Data__Chinook (**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__Sample_Data__Chinook(Kwargs_To_Self):
    url_chinook_database_json : str                 = URL__CHINOOK_JSON
    path_local_db             : str                 = PATH__DB__CHINOOK
    table_name                : str                 = TABLE_NAME__CHINOOK
    json_db                   : Sqlite__DB__Json



    def chinook_data_as_json(self):
        path_chinook_data_as_json = self.path_chinook_data_as_json()
        if file_not_exists(path_chinook_data_as_json):
            GET_to_file(self.url_chinook_database_json, path_chinook_data_as_json)
        return json_from_file(path_chinook_data_as_json)

    # def create_table_from_data(self):
    #     chinook_data = self.chinook_data_as_json()
    #     table_creator = Sqlite__Table__Create(table_name=self.table_name)
    #     table         = table_creator.table
    #     table_creator.add_fields__text("name", "value").create_table()
    #
    #     cursor = table.cursor()
    #     assert len(chinook_data) == 11
    #     for key, items in chinook_data.items():
    #         name = key
    #         value = json_dump(items)
    #         table.row_add(dict(name=name, value=value))
    #
    #     cursor.commit()
    #     return table

    def create_tables(self):
        self.create_tables_from_schema()
        chinook_data = self.chinook_data_as_json()
        for table_name, rows in chinook_data.items():
            table = self.json_db.database.table(table_name)
            table.rows_add(rows)



    def create_tables_from_schema(self):
        schemas  = self.tables_schemas_fields_from_data()
        for table_name, table_schema in schemas.items():
            self.json_db.create_table_from_schema(table_name, table_schema)
        return self

    def database(self):                         # todo: need to refactor this code based from the creation of the database to the use of it
        return self.json_db.database

    @cache_on_self
    def load_db_from_disk(self):
        db_chinook = Sqlite__Database(db_path=PATH__DB__CHINOOK, auto_schema_row=True)          # set the auto_schema_row so that we have a row_schema defined for all tables
        return db_chinook
        # db_chinook.connect()
        # return db_chinook.table(self.table_name)

    def path_chinook_data_as_json(self):
        return path_combine(self.path_chinook_data(), FILE_NAME__CHINOOK_DATA_JSON)

    def path_chinook_data(self):
        path_chinook_data = path_combine(self.path_sqlite_sample_data_sets(), FOLDER_NAME__CHINOOK_DATA)
        return folder_create(path_chinook_data)

    def path_sqlite_sample_data_sets(self):                     # todo: refactor to sqlite_sample_data_sets helper class
        path_data_sets = path_combine(current_temp_folder(), FOLDER_NAME__SQLITE_DATA_SETS)
        return folder_create(path_data_sets)

    def path_tables_schemas_fields(self):
        return path_combine(self.path_chinook_data(), FILE_NAME__TABLES_SCHEMAS_FIELDS)

    def table__genre(self):
        return self.load_db_from_disk().table('Genre')          # todo: refactor to sqlite_sample_data_sets helper class

    def table__track(self):
        return self.load_db_from_disk().table('Track')          # todo: refactor to sqlite_sample_data_sets helper class

    # def tables(self):
    #     return self.load_db_from_disk().tables()

    def tables_schemas_fields_from_data(self):
        path_tables_schemas_fields = self.path_tables_schemas_fields()
        if file_not_exists(path_tables_schemas_fields):
            tables_schemas = {}
            for name, data  in self.chinook_data_as_json().items():
                table_schema = self.json_db.get_schema_from_json_data(data)
                tables_schemas[name] = table_schema
            json_file_create(tables_schemas, path=path_tables_schemas_fields)

        return json_file_load(path_tables_schemas_fields)

    def save(self, path=PATH__DB__CHINOOK):
        database = self.database()
        database.save_to(path)
        return True

Ancestors

Class variables

var json_dbSqlite__DB__Json
var path_local_db : str
var table_name : str
var url_chinook_database_json : str

Methods

def chinook_data_as_json(self)
Expand source code
def chinook_data_as_json(self):
    path_chinook_data_as_json = self.path_chinook_data_as_json()
    if file_not_exists(path_chinook_data_as_json):
        GET_to_file(self.url_chinook_database_json, path_chinook_data_as_json)
    return json_from_file(path_chinook_data_as_json)
def create_tables(self)
Expand source code
def create_tables(self):
    self.create_tables_from_schema()
    chinook_data = self.chinook_data_as_json()
    for table_name, rows in chinook_data.items():
        table = self.json_db.database.table(table_name)
        table.rows_add(rows)
def create_tables_from_schema(self)
Expand source code
def create_tables_from_schema(self):
    schemas  = self.tables_schemas_fields_from_data()
    for table_name, table_schema in schemas.items():
        self.json_db.create_table_from_schema(table_name, table_schema)
    return self
def database(self)
Expand source code
def database(self):                         # todo: need to refactor this code based from the creation of the database to the use of it
    return self.json_db.database
def load_db_from_disk(self)
Expand source code
@cache_on_self
def load_db_from_disk(self):
    db_chinook = Sqlite__Database(db_path=PATH__DB__CHINOOK, auto_schema_row=True)          # set the auto_schema_row so that we have a row_schema defined for all tables
    return db_chinook
    # db_chinook.connect()
    # return db_chinook.table(self.table_name)
def path_chinook_data(self)
Expand source code
def path_chinook_data(self):
    path_chinook_data = path_combine(self.path_sqlite_sample_data_sets(), FOLDER_NAME__CHINOOK_DATA)
    return folder_create(path_chinook_data)
def path_chinook_data_as_json(self)
Expand source code
def path_chinook_data_as_json(self):
    return path_combine(self.path_chinook_data(), FILE_NAME__CHINOOK_DATA_JSON)
def path_sqlite_sample_data_sets(self)
Expand source code
def path_sqlite_sample_data_sets(self):                     # todo: refactor to sqlite_sample_data_sets helper class
    path_data_sets = path_combine(current_temp_folder(), FOLDER_NAME__SQLITE_DATA_SETS)
    return folder_create(path_data_sets)
def path_tables_schemas_fields(self)
Expand source code
def path_tables_schemas_fields(self):
    return path_combine(self.path_chinook_data(), FILE_NAME__TABLES_SCHEMAS_FIELDS)
def save(self, path='/tmp/db-tests/test-chinook.db')
Expand source code
def save(self, path=PATH__DB__CHINOOK):
    database = self.database()
    database.save_to(path)
    return True
def table__genre(self)
Expand source code
def table__genre(self):
    return self.load_db_from_disk().table('Genre')          # todo: refactor to sqlite_sample_data_sets helper class
def table__track(self)
Expand source code
def table__track(self):
    return self.load_db_from_disk().table('Track')          # todo: refactor to sqlite_sample_data_sets helper class
def tables_schemas_fields_from_data(self)
Expand source code
def tables_schemas_fields_from_data(self):
    path_tables_schemas_fields = self.path_tables_schemas_fields()
    if file_not_exists(path_tables_schemas_fields):
        tables_schemas = {}
        for name, data  in self.chinook_data_as_json().items():
            table_schema = self.json_db.get_schema_from_json_data(data)
            tables_schemas[name] = table_schema
        json_file_create(tables_schemas, path=path_tables_schemas_fields)

    return json_file_load(path_tables_schemas_fields)

Inherited members