diff --git a/src/byteb4rb1e/utils/dataclasses/__init__.py b/src/byteb4rb1e/utils/dataclasses/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/byteb4rb1e/utils/dataclasses/sql.py b/src/byteb4rb1e/utils/dataclasses/sql.py new file mode 100644 index 0000000..9f49477 --- /dev/null +++ b/src/byteb4rb1e/utils/dataclasses/sql.py @@ -0,0 +1,147 @@ +from dataclasses import dataclass, fields, Field +import datetime +from typing import Optional, get_origin, get_args, Union, List + + +def escape_value(value: str) -> str: + """Escapes a string value for safe SQL insertion. + + :param value: Raw string value + + :return: Escaped SQL-safe string + """ + return "'" + str(value).replace("'", "''") + "'" + + +def extract_fields_from_expr(expr: Expr) -> List[str]: + """Recursively extracts field names used in an expression tree. + + :param expr: Expression object + + :return: List of field names referenced in the expression + """ + if isinstance(expr, (Eq, Gt, Lt, Ne, Like)): + return [expr.field] + elif isinstance(expr, (And, Or)): + fields = [] + for sub in expr.conditions: + fields.extend(extract_fields_from_expr(sub)) + return fields + return [] + + +@dataclass(freeze=True) +class GIDModelProps: + """ + """ + table_prefix: Optional[str] = None, + pk_name: Optional[str] = None + + +class GIDModelParent: + """ + """ + table_name: str + foreign_key_name: str + + +@dataclass +class GIDModel: + """Base class for SQL-aware dataclasses with support for schema generation, + insertion, and query generation. + """ + + @classmethod + def create_table_sql( + cls: "GIDModel", + props: GIDModelProps, + parent: Optional[GIDModelParent] = None, + tables: Optional[[str, GIDModel]] = None, + ) -> str: + """Generates SQL CREATE TABLE statements for the model and its nested + components. + + :param table_prefix: Optional prefix for table names + :param root_table: Name of the root table for foreign key references + + :return: SQL CREATE TABLE statement(s) + """ + tables = tables or [] + + table_name = f"{{props.table_prefix} or ''}{cls.__name__.lower()}" + + if table_name in tables.keys(): + if not cls == tables[table_name]: + raise Exception('table with name '{table_name}' already incorporated with different schema') + else: + return None + + columns = ["{props.pk_name} TEXT PRIMARY KEY"] if props.pk_name else [] + + nested_sql = [] + foreign_keys = [] + + if parent and parent.table_name != table_name: + foreign_keys.append( + f"FOREIGN KEY (gid) REFERENCES {parent.table_name}({parent.foreign_key_name}) " + f"ON DELETE CASCADE ON UPDATE CASCADE" + ) + + for f in fields(cls): + if f.name == props.pk_name: continue # already handled + + py_type = f.type + + + if get_origin(py_type) is Union and get_args(py_type)[1] is type(None): + py_type = get_args(py_type)[0] + + sql_type = SQL_TYPECAST.get(py_type) + + metadata = f.metadata.get("gid", None) + + if issubclass(py_type, List): + py_type = get_args(py_type)[0] + pk_name = None + parent_fk_name = f.name + else: + pk_name = 'id' + parent_fk_name = f.name + + if issubclass(py_type, GIDModel): + if props.pk_name: + raise Exception('gidless model with nesting not possible') + nested_table = f"{props.table_prefix}{py_type.__name__.lower()}" + + # determine whether to pass parent or self + # if no primary key, pass the parent if it exists + if not props.pk_name: + f_parent=parent or None + # if primary, pass self + else: + f_parent=GIDModelParent( + table_name=table, + fk_name=f.name + ) + + nested_sql.append( + py_type.create_table_sql( + props=f_props or GIDModelProps( + table_name=f.name + pk_name=f_parent.fk_name + ) + parent=f_parent + ) + ) + if sql_type: + column_def = f"{f.name} {sql_type}" + if metadata: + column_def += f" {metadata}" + columns.append(column_def) + else: + raise Exception(f"Unable to typecast field '{f.name}' with type '{py_type}' to SQL") + + all_defs = columns + foreign_keys + main_sql = f"CREATE TABLE {table} (\n {',\n '.join(all_defs)}\n);" + return "\n".join([main_sql] + nested_sql) + diff --git a/tests/unit/byteb4rb1e/utils/dataclasses/__init__.py b/tests/unit/byteb4rb1e/utils/dataclasses/__init__.py new file mode 100644 index 0000000..e69de29