Initial attempt at a static version of df.py

Author: Tim Band

datafaker/populate.py

@@ -0,0 +1,189 @@
+"""This file was auto-generated by datafaker but can be edited manually."""
+from collections.abc import Iterable, Mapping, MutableMapping, Sequence
+from mimesis import Generic, Numeric, Person
+from mimesis.locales import Locale
+import sqlalchemy
+import sys
+from typing import Any, Callable
+import yaml
+
+from datafaker.base import FileUploader, TableGenerator, DistributionGenerator, ColumnPresence
+from datafaker.main import load_metadata
+from datafaker.make import TableGeneratorInfo, StoryGeneratorInfo  #TODO: move these in here!
+
+from datafaker.providers import (
+    BytesProvider,
+    ColumnValueProvider,
+    NullProvider,
+    SQLGroupByProvider,
+    TimedeltaProvider,
+    TimespanProvider,
+    WeightedBooleanProvider,
+)
+from datafaker.utils import logging, get_vocabulary_table_names
+
+generic = Generic(locale=Locale.EN_GB)
+numeric = Numeric()
+person = Person()
+dist_gen = DistributionGenerator()
+column_presence = ColumnPresence()
+
+generic.add_provider(BytesProvider)
+generic.add_provider(ColumnValueProvider)
+generic.add_provider(NullProvider)
+generic.add_provider(SQLGroupByProvider)
+generic.add_provider(TimedeltaProvider)
+generic.add_provider(TimespanProvider)
+generic.add_provider(WeightedBooleanProvider)
+
+#metadata = load_metadata("{{ orm_file_name }}", "{{ config_file_name }}")
+
+#import {{ row_generator_module_name }}
+#import {{ story_generator_module_name }}
+
+def _eval_structure(config: Any, context: Mapping) -> Any:
+    """
+    Turn a structure from ``config.yaml`` into a Python object.
+
+    :param config: a structure (list, dict, number or expression in a string).
+    :return: Object matching the structure of ``config`` with strings eval'ed.
+    """
+    if isinstance(config, str):
+        return eval(config, locals=context)
+    if isinstance(config, Mapping):
+        return {
+            k: _eval_structure(v, context)
+            for k, v in config.items()
+        }
+    if isinstance(config, Sequence):
+        return [_eval_structure(v, context) for v in config]
+    return config
+
+
+def _get_object(class_name: Any, context: Mapping) -> Any:
+    """
+    Get an object out of the context.
+
+    :param class_name: The name of the class, qualified if necessary.
+    Like "module.MyClass.Nested"
+    :param context: Mapping of strings to objects with those names.
+    :return: A value from ``context`` if there are no qualifying names,
+    otherwise the attribute of the base object.
+    """
+    if not isinstance(class_name, str):
+        return None
+    if not isinstance(kwargs, Mapping):
+        kwargs = {}
+    parts = class_name.split(".")
+    if parts[0] not in context:
+        logging.error('No such object "%"', parts[0])
+        return None
+    value = context[parts[0]]
+    so_far = parts[0]
+    for part in parts[1:]:
+        so_far += "." + part
+        if not hasattr(value, part):
+            logging.error('No such attribute "%"', so_far)
+            return None
+        value = getattr(value, part)
+    return value
+
+
+def _call_from_context(callable_name: Any, kwargs: Any, context: Mapping) -> Any:
+    """
+    Call a callable from the classes (or functions) in the context.
+
+    :param class_name: Possibly qualified name of class to construct.
+    :param context: Mapping of base classes and modules
+    :return: Constructed object, or None if this did not work.
+    """
+    cls = _get_object(callable_name, context)
+    if not isinstance(cls, Callable):
+        return None
+    kws = _eval_structure(kwargs, context)
+    if kws is None:
+        return None
+    return cls(**kws)
+
+
+def _get_src_stats(src_stats_filename: str) -> Any:
+    """
+    Get the SRC_STATS object
+    """
+    with open("{{ src_stats_filename }}", "r", encoding="utf-8") as f:
+        return yaml.unsafe_load(f)
+
+
+class TableGenerator:
+
+    def __init__(self, rows_per_pass: int, dst_db_conn: sqlalchemy.Connection, table_data: TableGeneratorInfo, max_unique_constraint_tries: int | None):
+        self.num_rows_per_pass = rows_per_pass
+        self.table_data = table_data
+        self.max_unique_constraint_tries = max_unique_constraint_tries
+        self.existing_constraint_hashes: MutableMapping[str, set[int]] = {}
+        self.context: Mapping = {}
+        for constraint in table_data.unique_constraints:
+            expr = sqlalchemy.select(constraint.columns)
+            query_result = dst_db_conn.execute(expr).fetchall()
+            self.existing_constraint_hashes[constraint.name] = set([
+                hash(tuple(result))
+                for result in query_result
+            ])
+
+    def set_context(self, context: Mapping):
+        self.context = context
+
+    def __call__(self, dst_db_conn):
+        result = {}
+        columns_to_generate = set(self.table_data.nonnull_columns)
+        # Which missingness patterns do we want?
+        for choice in self.table_data.column_choices:
+            cols = _call_from_context(choice.function_name, choice.argument_values, self.context)
+            columns_to_generate.update(cols)
+
+        max_tries = self.max_unique_constraint_tries
+        while columns_to_generate:
+            if max_tries == 0:
+                raise RuntimeError(f"Failed to satisfy unique constraints for table {self.table_data.table_name} after {self.max_unique_constraint_tries} attempts.")
+            if max_tries is not None:
+                max_tries -= 1
+            for row_gen in self.table_data.row_gens:
+                if set(row_gen.variable_names) & columns_to_generate:
+                    values = _call_from_context(row_gen.function_call.function_name, row_gen.function_call.argument_values, self.context)
+                    for index, variable_name in enumerate(row_gen.variable_names):
+                        result[variable_name] = values[index]
+            columns_to_generate = set()
+            for constraint in self.table_data.unique_constraints:
+                cf_hash = hash(tuple(
+                    result[col.name] for col in constraint.columns
+                ))
+                if cf_hash in self.existing_constraint_hashes[constraint.name]:
+                    columns_to_generate.update(c.name for c in constraint.columns)
+        for constraint in self.table_data.unique_constraints:
+            cf_hash = hash(tuple(
+                result[col.name] for col in constraint.columns
+            ))
+            self.existing_constraint_hashes.add(cf_hash)
+        return result
+
+def get_table_generator_dict(self, rows_per_pass: int, dst_db_conn: sqlalchemy.Connection, tables_data: Iterable[TableGeneratorInfo], max_unique_constraint_tries: int | None):
+    return {
+    "{{ table_data.table_name }}": TableGenerator(rows_per_pass, dst_db_conn, table_data, max_unique_constraint_tries)
+    for table_data in tables_data
+}
+
+
+def get_vocab_dict(config: Mapping, metadata: sqlalchemy.MetaData) -> Mapping[str, FileUploader]: {
+    name: FileUploader[metadata.tables[name]]
+    for name in get_vocabulary_table_names(config)
+}
+
+def get_story_generator_list(story_generator_infos: Iterable[StoryGeneratorInfo], context: Mapping) -> list[Mapping]:
+    return [
+        {
+            "function": _call_from_context(gen_data.function_call.function_name, gen_data.function_call.argument_values, context),
+            "num_stories_per_pass": {{ gen_data.num_stories_per_pass }},
+            "name": "{{ gen_data.function_call.function_name }}",
+        }
+        for gen_data in story_generator_infos
+    ]