Reference¶

class DataGenerator:
  """
  Generate pandas DataFrames, dict streams or files from a RandSpec.

  ``stream_dict`` converts normalized microbatches to pandas records. Text columns of ``get_df``
  are ``string[pyarrow]`` with ``pd.NA`` nulls; a transformer's output keeps the dtype it returns; ``stream_dict`` records hold Python ``None`` for nulls.

  Parameters
  ----------
  random_spec : dict or callable
      RandSpec mapping each column to ``{"method": ..., "kwargs": ...}``, or a
      zero-argument callable returning one. Validated at construction.
  seed : int, optional
      Seed of this generator's own random state; the same spec, seed and size
      give the same output.

  Raises
  ------
  SpecValidationError
      If the evaluated spec is invalid.

  Examples
  --------
  >>> from rand_engine import DataGenerator
  >>> spec = {
  ...     "user_id": {"method": "int_zfilled", "kwargs": {"length": 8}},
  ...     "age": {"method": "integers", "kwargs": {"min": 18, "max": 80}},
  ... }
  >>> df = DataGenerator(spec, seed=42).size(100).get_df()
  >>> list(df.columns)
  ['user_id', 'age']
  >>> bool(df["age"].between(18, 80).all())
  True
  """

  def __init__(self, random_spec: Callable[[], dict] | dict, seed: int = None):
    self.lazy_random_spec = random_spec
    self.__validate_spec()

    seed_sequence = np.random.SeedSequence(seed)
    self._rng = np.random.default_rng(seed_sequence)
    self._key_seed = int(seed_sequence.generate_state(1)[0])
    self._size = None
    self._transformers: List[Callable] = []


  def __evaluate_spec(self):
    if callable(self.lazy_random_spec): 
      return self.lazy_random_spec()
    return self.lazy_random_spec


  def __validate_spec(self):
    evaluated_spec = self.__evaluate_spec()
    AdvancedValidator.validate_and_raise(evaluated_spec)


  def _file_seam(self) -> tuple[Callable[..., pa.Schema], Callable[[int, int], pd.DataFrame]]:
    """Evaluate the spec once; return its catalog schema and a batch factory over that mapping."""
    evaluated_spec = self.__evaluate_spec()
    rand_generator = RandGenerator(evaluated_spec)

    def batch(size: int, offset: int = 0) -> pd.DataFrame:
      frame = rand_generator.generate_first_level(size=size, rng=self._rng, key_seed=self._key_seed, offset=offset)
      frame = rand_generator.apply_embedded_transformers(frame)
      frame = rand_generator.apply_global_transformers(frame, self._transformers)
      if not isinstance(frame, pd.DataFrame) or len(frame.index) != size:
        actual = len(frame.index) if isinstance(frame, pd.DataFrame) else "non-DataFrame"
        raise RandEngineError(f"global transformers must preserve row count {size}; got {actual}")
      return rand_generator.apply_modifiers(frame, self._rng)

    return cache(partial(self._catalog_schema, evaluated_spec)), batch


  def wrapped_df_generator(
    self, size: int, offset: int = 0
  ) -> Callable[[], pd.DataFrame]:
    """
    Return a zero-argument callable that generates one batch of ``size`` rows.

    The callable runs generation, embedded transformers, global transformers and
    modifiers, in that order.

    Parameters
    ----------
    size : int
        Rows in the batch.
    offset : int, default 0
        Row index of the batch's first row, so streamed and multi-part rows
        continue one index.

    Returns
    -------
    callable
        ``() -> pandas.DataFrame``; each call generates a new batch.

    Raises
    ------
    RandEngineError
        When called, if a global transformer changes the row count.

    Examples
    --------
    >>> from rand_engine import DataGenerator
    >>> spec = {"age": {"method": "integers", "kwargs": {"min": 18, "max": 80}}}
    >>> batch = DataGenerator(spec, seed=42).wrapped_df_generator(size=4)
    >>> batch().shape
    (4, 1)
    """
    return lambda: self._file_seam()[1](size, offset)


  def transformers(self, transformers: List[Callable]):
    """
    Set the functions that transform each generated batch, in order.

    Parameters
    ----------
    transformers : list of callable
        Functions ``DataFrame -> DataFrame``; each must keep the batch's row count.

    Returns
    -------
    DataGenerator
        This generator, for chaining.

    Raises
    ------
    RandEngineError
        When a batch is generated, if a transformer changes its row count.

    Examples
    --------
    >>> from rand_engine import DataGenerator
    >>> spec = {"age": {"method": "integers", "kwargs": {"min": 18, "max": 80}}}
    >>> gen = DataGenerator(spec, seed=42).transformers([lambda df: df.assign(double=df["age"] * 2)]).size(10)
    >>> list(gen.get_df().columns)
    ['age', 'double']
    """
    self._transformers = transformers
    return self


  def size(self, size: int | Callable[[], int]):
    """
    Set the rows per batch and return this generator.

    Parameters
    ----------
    size : int or callable
        Rows per ``get_df`` call, total rows of a ``write``, or rows per stream
        microbatch. A callable returns an int and is called once per ``get_df``,
        ``write.save`` or ``stream_dict`` call, and once per pass of ``writeStream``.

    Returns
    -------
    DataGenerator
        This generator, for chaining.

    Examples
    --------
    >>> from rand_engine import DataGenerator
    >>> spec = {"age": {"method": "integers", "kwargs": {"min": 18, "max": 80}}}
    >>> DataGenerator(spec, seed=42).size(lambda: 5).get_df().shape
    (5, 1)
    """
    self._size = size
    return self


  def _resolve_size(self) -> int:
    if self._size is None:
      raise RandEngineError("No size set: call .size(n) with an int or a callable returning one.")
    return self._size() if callable(self._size) else self._size


  def _catalog_schema(self, evaluated_spec: dict, exact: bool = False) -> pa.Schema:
    """Return the spec's declared columns as a schema; `exact` (an empty file) refuses when transformers can change it."""
    if exact and (self._transformers or any(config.get("transformers") for config in evaluated_spec.values())):
      raise RandEngineError("empty output schema is indeterminate when transformers are configured")
    fields = []
    try:
      for spec_name, config in evaluated_spec.items():
        names = config.get("cols", [spec_name])
        types = METHOD_CATALOG[config["method"]].arrow_types(config.get("kwargs", {}))
        if len(names) != len(types):
          raise RandEngineError(
            f"method '{config['method']}' declares {len(types)} output types for {len(names)} columns"
          )
        fields.extend(pa.field(name, type_, nullable=True) for name, type_ in zip(names, types))
    except RandEngineError:
      raise
    except Exception as error:
      raise RandEngineError(f"empty output schema cannot be derived: {type(error).__name__}") from error
    return pa.schema(fields)


  def get_df(self):
    """
    Return one DataFrame of ``size`` rows generated from the spec.

    Returns
    -------
    pandas.DataFrame
        ``size`` rows with the spec's columns, plus any columns the transformers add.

    Raises
    ------
    RandEngineError
        If ``size`` was not set.

    Examples
    --------
    >>> from rand_engine import DataGenerator, RandSpecs
    >>> df = DataGenerator(RandSpecs.customers(), seed=42).size(10).get_df()
    >>> len(df)
    10
    """
    lazy_dataframe = self.wrapped_df_generator(size=self._resolve_size())
    assert lazy_dataframe is not None, "You need to generate a DataFrame first."
    assert callable(lazy_dataframe), "wrapped_df_generator must return a callable"
    return lazy_dataframe()


  def stream_dict(self, min_throughput: int=1, max_throughput: int = 10) -> Generator:
    """
    Yield the rows of an endless stream as dicts, one microbatch of ``size`` rows at a time.

    Each record gains a ``timestamp_created`` key (epoch seconds, 3 decimals); text columns,
    dates included, are ``str`` and nulls are ``None``. After each record the generator sleeps
    ``1 / uniform(min_throughput, max_throughput)`` seconds.

    Parameters
    ----------
    min_throughput : int, default 1
        Lower bound of the uniform draw that sets the sleep after each record.
    max_throughput : int, default 10
        Upper bound of the uniform draw that sets the sleep after each record.

    Yields
    ------
    dict
        One row of the spec's columns plus ``timestamp_created``; nulls are ``None``.

    Raises
    ------
    RandEngineError
        On the first ``next()``, if ``size`` was not set.

    Examples
    --------
    >>> from itertools import islice
    >>> from rand_engine import DataGenerator
    >>> spec = {"age": {"method": "integers", "kwargs": {"min": 18, "max": 80}}}
    >>> stream = DataGenerator(spec, seed=42).size(5).stream_dict(min_throughput=500, max_throughput=1000)
    >>> sorted(next(stream))
    ['age', 'timestamp_created']
    """
    size, offset = self._resolve_size(), 0
    while True:
      df_data_microbatch = self.wrapped_df_generator(size, offset)()
      offset += size
      df_data_parsed = StreamHandler.convert_dt_to_str(df_data_microbatch)
      list_of_records = df_data_parsed.to_dict('records')
      for record in list_of_records:
        record["timestamp_created"] = round(time.time(), 3)
        yield record
        StreamHandler.sleep_to_contro_throughput(min_throughput, max_throughput)


  @property
  def write(self):
    """
    Return the batch file writer of this generator.

    Chain ``format``, ``mode``, ``option`` or ``options``, then call ``save(path)``:
    it generates ``size`` rows and writes them to ``path`` (one file, or ``numFiles``
    parts). Nothing is written until ``save`` runs.

    Returns
    -------
    FileBatchWriter
        Writer whose ``save(path)`` writes files at the relative or absolute ``path``.

    Examples
    --------
    >>> import os, tempfile
    >>> from rand_engine import DataGenerator
    >>> spec = {"age": {"method": "integers", "kwargs": {"min": 18, "max": 80}}}
    >>> out = tempfile.mkdtemp()
    >>> DataGenerator(spec, seed=42).size(10).write.format("csv").save(os.path.join(out, "ages.csv"))
    >>> os.path.isfile(os.path.join(out, "ages.csv"))
    True
    """
    return FileBatchWriter(self._resolve_size, self.wrapped_df_generator, self._file_seam)


  @property
  def writeStream(self):
    """
    Return the stream file writer of this generator.

    Chain ``format``, ``trigger(frequency)`` and ``option("timeout", seconds)`` (default
    20), then call ``start(path)``: each pass writes one microbatch of ``size`` rows to
    ``path/part-<uuid>.<ext>``, sleeps ``frequency`` seconds, and repeats until
    ``timeout`` seconds have passed. ``start`` blocks for that duration.

    Returns
    -------
    FileStreamWriter
        Writer whose ``start(path)`` writes files under ``path`` and returns ``None``.

    Examples
    --------
    >>> import os, tempfile
    >>> from rand_engine import DataGenerator
    >>> spec = {"age": {"method": "integers", "kwargs": {"min": 18, "max": 80}}}
    >>> out = tempfile.mkdtemp()
    >>> target = os.path.join(out, "ages")
    >>> stream = DataGenerator(spec, seed=42).size(10).writeStream
    >>> stream.format("json").trigger(frequency=0.1).option("timeout", 0.3).start(target)
    >>> bool(os.listdir(target))
    True
    """
    return FileStreamWriter(self._resolve_size, self.wrapped_df_generator)

class SparkGenerator:
  """
  Generate a PySpark DataFrame from a RandSpec.

  Ordinary columns use native Spark expressions. Stateless ``pk`` and ``fk`` columns use
  the shared key authority; an explicit seed applies to those key columns only.

  Parameters
  ----------
  spark : pyspark.sql.SparkSession
      Session that builds the DataFrame.
  F : module
      The ``pyspark.sql.functions`` module.
  metadata : dict
      RandSpec mapping each column to ``{"method": ..., "kwargs": ...}``. Validated at construction.

  Raises
  ------
  SpecValidationError
      If ``metadata`` is invalid.

  Examples
  --------
  >>> from pyspark.sql import SparkSession, functions as F  # doctest: +SKIP
  >>> from rand_engine import SparkGenerator  # doctest: +SKIP
  >>> spark = SparkSession.builder.getOrCreate()  # doctest: +SKIP
  >>> spec = {"age": {"method": "integers", "kwargs": {"min": 18, "max": 80}}}
  >>> SparkGenerator(spark, F, spec).size(100).get_df().count()  # doctest: +SKIP
  100
  """

  def __init__(self, spark, F, metadata, seed=None):
    # Validação obrigatória - sempre executada
    key_metadata = {}
    if isinstance(metadata, dict):
      key_metadata = {
        column: config for column, config in metadata.items()
        if isinstance(config, dict) and config.get("method") in ("pk", "fk")
      }
    if key_metadata:
      ordinary_metadata = {
        column: config for column, config in metadata.items()
        if column not in key_metadata
      }
      errors = (
        CommonValidator.validate_spark_spec(ordinary_metadata)
        if ordinary_metadata else []
      )
      errors.extend(AdvancedValidator.validate(key_metadata))
    else:
      errors = CommonValidator.validate_spark_spec(metadata)
    if errors:
      CommonValidator._raise(errors, "SPARKGENERATOR")

    self.spark = spark
    self.F = F
    self.metadata = metadata
    self._key_metadata = key_metadata
    if key_metadata:
      seed_sequence = np.random.SeedSequence(seed)
      self._key_seed = int(seed_sequence.generate_state(1)[0])
    self._size = None

  def map_methods(self):
    """
    Return the RandSpec method names mapped to their Spark generator functions.

    Returns
    -------
    dict
        Method name to its Spark or stateless key generator function.

    Examples
    --------
    >>> from pyspark.sql import SparkSession, functions as F  # doctest: +SKIP
    >>> from rand_engine import SparkGenerator  # doctest: +SKIP
    >>> spark = SparkSession.builder.getOrCreate()  # doctest: +SKIP
    >>> spec = {"age": {"method": "integers", "kwargs": {"min": 18, "max": 80}}}  # doctest: +SKIP
    >>> "integers" in SparkGenerator(spark, F, spec).map_methods()  # doctest: +SKIP
    True
    """
    return {
      "integers": SparkCore.gen_ints,
      "int_zfilled": SparkCore.gen_ints_zfilled,
      "floats": SparkCore.gen_floats,
      "floats_normal": SparkCore.gen_floats_normal,
      "distincts": SparkCore.gen_distincts,
      "distincts_prop": SparkCore.gen_distincts_prop,
      "unix_timestamps": SparkCore.gen_unix_timestamps,
      "uuid4": SparkCore.gen_uuid4,
      "booleans": SparkCore.gen_booleans,
      "dates": SparkCore.gen_dates,
      "pk": Keys.gen_pk,
      "fk": Keys.gen_fk,
    }

  def size(self, size):
    """
    Set the number of rows of the DataFrame and return this generator.

    Parameters
    ----------
    size : int
        Rows built by ``get_df``.

    Returns
    -------
    SparkGenerator
        This generator, for chaining.

    Examples
    --------
    >>> from pyspark.sql import SparkSession, functions as F  # doctest: +SKIP
    >>> from rand_engine import SparkGenerator  # doctest: +SKIP
    >>> spark = SparkSession.builder.getOrCreate()  # doctest: +SKIP
    >>> spec = {"age": {"method": "integers", "kwargs": {"min": 18, "max": 80}}}  # doctest: +SKIP
    >>> gen = SparkGenerator(spark, F, spec).size(100)  # doctest: +SKIP
    """
    self._size = size
    return self


  def get_df(self):
    """
    Build the Spark DataFrame of ``size`` rows generated from the metadata.

    Returns
    -------
    pyspark.sql.DataFrame
        One column per metadata key; the technical ``id`` column is dropped unless the metadata declares ``id``.

    Raises
    ------
    RandEngineError
        If ``size`` was not set.

    Examples
    --------
    >>> from pyspark.sql import SparkSession, functions as F  # doctest: +SKIP
    >>> from rand_engine import SparkGenerator  # doctest: +SKIP
    >>> spark = SparkSession.builder.getOrCreate()  # doctest: +SKIP
    >>> spec = {"age": {"method": "integers", "kwargs": {"min": 18, "max": 80}}}
    >>> SparkGenerator(spark, F, spec).size(100).get_df().count()  # doctest: +SKIP
    100
    """
    if self._size is None:
      raise RandEngineError("size not set: call .size(n) before get_df()")
    mapped_methods = self.map_methods()
    dataframe = self.spark.range(self._size)
    source_index = next(
      f"__rand_engine_source_index_{index}"
      for index in range(len(self.metadata) + 1)
      if f"__rand_engine_source_index_{index}" not in self.metadata
    )
    if self._key_metadata:
      dataframe = dataframe.withColumn(source_index, self.F.col("id"))
    for k, v in self.metadata.items():
      if v["method"] in ("pk", "fk"):
        continue
      generator_method = mapped_methods[v["method"]]
      dataframe = generator_method(self.spark, F=self.F, df=dataframe, col_name=k, **v.get("kwargs", {}))
    if self._key_metadata:
      if "id" in self._key_metadata:
        dataframe = dataframe.drop("id")
      dataframe = self._add_key_columns(dataframe, source_index)
      return dataframe.select(*(
        self.F.col(f"`{column.replace('`', '``')}`").alias(column)
        for column in self.metadata
      ))
    # Remove the technical 'id' column from spark.range() only if not in spec
    if "id" not in self.metadata:
      return dataframe.drop("id")
    return dataframe


  def _add_key_columns(self, dataframe, source_index):
    """Build stateless key columns from the source indexes carried by Spark."""
    from pyspark.sql.types import LongType, StringType, StructField, StructType

    key_metadata = self._key_metadata
    key_seed = self._key_seed
    for column, config in key_metadata.items():
      if config["method"] != "fk" or self._size == 0:
        continue
      kwargs = config.get("kwargs", {})
      born_before = kwargs.get("born_before")
      if born_before is None:
        continue
      eligible = Keys._eligible_parent_counts(
        np.asarray([0], dtype=np.int64), kwargs["parent_size"], born_before
      )
      if not eligible[0]:
        raise RandEngineError(f"fk column '{column}' has a child with no eligible parent")

    def key_field(column, config):
      kwargs = config.get("kwargs", {})
      arrow_type = METHOD_CATALOG[config["method"]].arrow_types(kwargs)[0]
      return StructField(column, StringType() if pa.types.is_string(arrow_type) else LongType(), False)

    schema = StructType([
      StructField(source_index, LongType(), False),
      *(key_field(column, config) for column, config in key_metadata.items()),
    ])

    def generate_keys(batches):
      for batch in batches:
        indexes = batch[source_index].to_numpy(dtype=np.int64)
        values = {source_index: indexes}
        for column, config in key_metadata.items():
          kwargs = config.get("kwargs", {})
          if config["method"] == "pk":
            values[column] = Keys._pk_at(
              indexes, int(indexes.min()) if indexes.size else 0,
              int(indexes.max()) if indexes.size else -1, **kwargs
            )
          else:
            values[column] = Keys._fk_at(
              indexes, kwargs["parent"], kwargs["parent_size"], kwargs.get("skew", 0),
              key_seed, column, kwargs.get("born_before")
            )
        yield pd.DataFrame(values)

    keys = dataframe.select(source_index).mapInPandas(generate_keys, schema)
    return dataframe.join(keys, source_index)

class CommonRandSpecs:
    """
    Cross-compatible specification examples for DataGenerator and SparkGenerator.

    All specs use the unified API with date_format parameter.

    Examples
    --------
    >>> from rand_engine import DataGenerator, RandSpecs
    >>> df = DataGenerator(RandSpecs.customers(), seed=42).size(100).get_df()
    >>> list(df.columns)
    ['customer_id', 'age', 'city', 'total_spent', 'is_premium', 'registration_date']
    >>> len(df)
    100
    """

    @classmethod
    def faker_pool(
        cls,
        provider: str,
        locale: str = "en_US",
        pool_size: int = 100,
        seed: int | None = None,
    ) -> Dict[str, Any]:
        """
        Build one materialized distincts spec with a local Faker instance.

        Parameters
        ----------
        provider : str
            Faker provider method, such as ``"name"`` or ``"city"``.
        locale : str, default "en_US"
            Faker locale; an unsupported locale raises RandEngineError.
        pool_size : int, default 100
            Number of values to materialize; a positive integer.
        seed : int, optional
            Seed of the Faker instance; the same seed gives the same pool.

        Returns
        -------
        dict
            A ``distincts`` RandSpec whose ``distincts`` list holds the pool.

        Raises
        ------
        RandEngineError
            If ``pool_size`` is not a positive integer, Faker is not installed, the locale
            or provider is unsupported or not callable, the provider fails, or a value is
            not scalar.

        Examples
        --------
        >>> from rand_engine import RandSpecs
        >>> spec = RandSpecs.faker_pool("city", pool_size=5, seed=1)
        >>> spec["method"]
        'distincts'
        >>> len(spec["kwargs"]["distincts"])
        5
        """
        if isinstance(pool_size, bool) or not isinstance(pool_size, int) or pool_size < 1:
            raise RandEngineError("pool_size must be a positive integer")

        try:
            from faker import Faker
        except ImportError as error:
            raise RandEngineError(
                "Faker is an optional dependency; install rand-engine[faker]"
            ) from error

        try:
            faker = Faker(locale)
        except Exception as error:
            raise RandEngineError("locale is not supported by Faker") from error

        try:
            faker.seed_instance(seed)
            factory = getattr(faker, provider)
        except (AttributeError, TypeError, ValueError) as error:
            raise RandEngineError("provider is not supported by Faker") from error
        if not callable(factory):
            raise RandEngineError("provider is not callable")

        try:
            values = [factory() for _ in range(pool_size)]
        except Exception as error:
            raise RandEngineError("provider failed while constructing the pool") from error
        if any(not is_declared_scalar(value) for value in values):
            raise RandEngineError("provider outputs must be scalar values")

        return {"method": "distincts", "kwargs": {"distincts": values}}

    @classmethod
    def from_schema(
        cls,
        schema: pa.Schema,
        overrides: dict[str, dict[str, Any]] | None = None,
    ) -> Dict[str, Any]:
        """
        Build a validated RandSpec from Arrow field metadata only.

        Parameters
        ----------
        schema : pyarrow.Schema
            Arrow schema; each field's type picks a default generation recipe.
        overrides : dict, optional
            Per-field overrides keyed by field name. A ``method`` key replaces the recipe; ``kwargs`` merge into it.

        Returns
        -------
        dict
            RandSpec with one entry per schema field, validated before return.

        Raises
        ------
        SpecValidationError
            If ``schema`` is not a ``pyarrow.Schema``; ``overrides`` or one of its values is not
            a dict, or an override's ``kwargs`` is not a dict; the schema has duplicate fields;
            an override names a missing field; a field type has no default recipe and no
            override with ``method``; or the resulting spec fails validation.

        Examples
        --------
        >>> import pyarrow as pa
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> schema = pa.schema([("id", pa.int32()), ("active", pa.bool_())])
        >>> spec = RandSpecs.from_schema(schema, {"id": {"kwargs": {"max": 9}}})
        >>> {name: column["method"] for name, column in spec.items()}
        {'id': 'integers', 'active': 'booleans'}
        >>> DataGenerator(spec, seed=1).size(5).get_df().shape
        (5, 2)
        """
        if not isinstance(schema, pa.Schema):
            raise SpecValidationError("from_schema requires a pyarrow.Schema")
        if overrides is None:
            overrides = {}
        if not isinstance(overrides, dict):
            raise SpecValidationError("overrides must be a dictionary")
        if any(not isinstance(value, dict) for value in overrides.values()):
            raise SpecValidationError("each field override must be a dictionary")

        names = schema.names
        duplicate_names = sorted({name for name in names if names.count(name) > 1})
        if duplicate_names:
            details = ", ".join(
                f"{name} ({', '.join(str(field.type) for field in schema if field.name == name)})"
                for name in duplicate_names
            )
            raise SpecValidationError(
                f"duplicate schema fields: {details}; rename or remove each duplicate"
            )

        unknown = sorted(set(overrides) - set(names))
        if unknown:
            raise SpecValidationError(
                f"override names missing schema fields: {', '.join(unknown)}"
            )

        copied_overrides = copy.deepcopy(overrides)
        result: Dict[str, Any] = {}
        for field in schema:
            override = copied_overrides.get(field.name, {})
            if "method" in override:
                result[field.name] = override
                continue

            recipe = cls._schema_recipe(field.type)
            if recipe is None:
                raise SpecValidationError(
                    f"field '{field.name}' type '{field.type}' requires a complete override with method"
                )

            column = copy.deepcopy(recipe)
            override_kwargs = override.get("kwargs", {})
            if not isinstance(override_kwargs, dict):
                raise SpecValidationError(
                    f"field '{field.name}' override kwargs must be a dictionary"
                )
            column["kwargs"].update(override_kwargs)
            column.update({key: value for key, value in override.items() if key != "kwargs"})
            result[field.name] = column

        AdvancedValidator.validate_and_raise(result)
        return result

    @staticmethod
    def _schema_recipe(arrow_type: pa.DataType) -> Dict[str, Any] | None:
        if pa.types.is_boolean(arrow_type):
            return {"method": "booleans", "kwargs": {"true_prob": 0.5}}
        if pa.types.is_integer(arrow_type):
            return {
                "method": "integers",
                "kwargs": {
                    "min": 0,
                    "max": min(100, 2 ** arrow_type.bit_width - 1),
                    "int_type": str(arrow_type),
                },
            }
        if pa.types.is_floating(arrow_type):
            return {
                "method": "floats",
                "kwargs": {"min": 0, "max": 1, "decimals": 2},
            }
        if pa.types.is_string(arrow_type) or pa.types.is_large_string(arrow_type):
            return {"method": "uuid4", "kwargs": {}}
        if pa.types.is_date(arrow_type):
            return {
                "method": "dates",
                "kwargs": {
                    "start": "2000-01-01",
                    "end": "2030-12-31",
                    "date_format": "%Y-%m-%d",
                },
            }
        if pa.types.is_timestamp(arrow_type) and arrow_type.tz is None:
            return {
                "method": "dates",
                "kwargs": {
                    "start": "2000-01-01",
                    "end": "2030-12-31",
                    "date_format": "%Y-%m-%d %H:%M:%S",
                },
            }
        if pa.types.is_null(arrow_type):
            return {"method": "constant", "kwargs": {"value": None}}
        return None

    @classmethod
    def customers(cls) -> Dict[str, Any]:
        """
        Return the customer profiles RandSpec (6 Fields).

        Returns
        -------
        dict
            RandSpec mapping each of the 6 fields to its method and kwargs.

        Examples
        --------
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> list(DataGenerator(RandSpecs.customers(), seed=1).size(5).get_df().columns)
        ['customer_id', 'age', 'city', 'total_spent', 'is_premium', 'registration_date']
        """
        return {
            "customer_id": {"method": "uuid4", "kwargs": {}},
            "age": {"method": "integers", "kwargs": {"min": 18, "max": 80, "int_type": "int32"}},
            "city": {
                "method": "distincts",
                "kwargs": {
                    "distincts": ["São Paulo", "Rio de Janeiro", "Belo Horizonte",
                                "Salvador", "Brasília", "Curitiba", "Porto Alegre"]
                }
            },
            "total_spent": {"method": "floats_normal", "kwargs": {"mean": 1500.0, "std": 500.0, "decimals": 2}},
            "is_premium": {"method": "booleans", "kwargs": {"true_prob": 0.15}},
            "registration_date": {
                "method": "dates",
                "kwargs": {"start": "2020-01-01", "end": "2025-10-30", "date_format": "%Y-%m-%d"}
            }
        }

    @classmethod
    def products(cls) -> Dict[str, Any]:
        """
        Return the product catalog RandSpec (7 Fields).

        Returns
        -------
        dict
            RandSpec mapping each of the 7 fields to its method and kwargs.

        Examples
        --------
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> list(DataGenerator(RandSpecs.products(), seed=1).size(5).get_df().columns)
        ['product_id', 'sku', 'name', 'price', 'stock_quantity', 'category', 'is_active']
        """
        return {
            "product_id": {"method": "uuid4", "kwargs": {}},
            "sku": {"method": "int_zfilled", "kwargs": {"length": 8}},
            "name": {
                "method": "distincts",
                "kwargs": {
                    "distincts": ["Laptop Pro", "Wireless Mouse", "USB-C Cable",
                                "Mechanical Keyboard", "Monitor 27inch", "Webcam HD",
                                "Headset Gaming", "SSD 1TB", "RAM 16GB", "Charger USB"]
                }
            },
            "price": {"method": "floats", "kwargs": {"min": 9.99, "max": 2999.99, "decimals": 2}},
            "stock_quantity": {"method": "integers", "kwargs": {"min": 0, "max": 500, "int_type": "int32"}},
            "category": {
                "method": "distincts_prop",
                "kwargs": {
                    "distincts": {"Electronics": 40, "Accessories": 30, "Computers": 20, "Peripherals": 10}
                }
            },
            "is_active": {"method": "booleans", "kwargs": {"true_prob": 0.85}}
        }

    @classmethod
    def orders(cls) -> Dict[str, Any]:
        """
        Return the e-commerce orders RandSpec (6 Fields).

        Returns
        -------
        dict
            RandSpec mapping each of the 6 fields to its method and kwargs.

        Examples
        --------
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> list(DataGenerator(RandSpecs.orders(), seed=1).size(5).get_df().columns)
        ['order_id', 'customer_id', 'amount', 'order_timestamp', 'status', 'payment_method']
        """
        return {
            "order_id": {"method": "uuid4", "kwargs": {}},
            "customer_id": {"method": "uuid4", "kwargs": {}},
            "amount": {"method": "floats_normal", "kwargs": {"mean": 200.0, "std": 100.0, "decimals": 2}},
            "order_timestamp": {
                "method": "unix_timestamps",
                "kwargs": {"start": "2025-01-01", "end": "2025-10-30", "date_format": "%Y-%m-%d"}
            },
            "status": {
                "method": "distincts_prop",
                "kwargs": {"distincts": {"completed": 70, "pending": 20, "cancelled": 10}}
            },
            "payment_method": {
                "method": "distincts",
                "kwargs": {"distincts": ["credit_card", "debit_card", "pix", "boleto"]}
            }
        }

    @classmethod
    def transactions(cls) -> Dict[str, Any]:
        """
        Return the financial transactions RandSpec (7 Fields).

        Returns
        -------
        dict
            RandSpec mapping each of the 7 fields to its method and kwargs.

        Examples
        --------
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> list(DataGenerator(RandSpecs.transactions(), seed=1).size(5).get_df().columns)
        ['transaction_id', 'account_id', 'amount', 'transaction_type', 'timestamp', 'is_approved', 'fee']
        """
        return {
            "transaction_id": {"method": "uuid4", "kwargs": {}},
            "account_id": {"method": "uuid4", "kwargs": {}},
            "amount": {"method": "floats", "kwargs": {"min": 10.0, "max": 5000.0, "decimals": 2}},
            "transaction_type": {
                "method": "distincts",
                "kwargs": {"distincts": ["deposit", "withdrawal", "transfer", "payment"]}
            },
            "timestamp": {
                "method": "unix_timestamps",
                "kwargs": {"start": "2025-01-01", "end": "2025-10-30", "date_format": "%Y-%m-%d"}
            },
            "is_approved": {"method": "booleans", "kwargs": {"true_prob": 0.90}},
            "fee": {"method": "floats", "kwargs": {"min": 0.0, "max": 50.0, "decimals": 2}}
        }

    @classmethod
    def employees(cls) -> Dict[str, Any]:
        """
        Return the employee records RandSpec (8 Fields).

        Returns
        -------
        dict
            RandSpec mapping each of the 8 fields to its method and kwargs.

        Examples
        --------
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> list(DataGenerator(RandSpecs.employees(), seed=1).size(5).get_df().columns)
        ['employee_id', 'department', 'position', 'salary', 'hire_date', 'age', 'is_remote', 'performance_score']
        """
        return {
            "employee_id": {"method": "uuid4", "kwargs": {}},
            "department": {
                "method": "distincts_prop",
                "kwargs": {"distincts": {"Engineering": 50, "Product": 20, "Data": 15, "Operations": 15}}
            },
            "position": {
                "method": "distincts",
                "kwargs": {
                    "distincts": ["Software Engineer", "Data Analyst", "Product Manager",
                                "DevOps Engineer", "QA Engineer", "Designer"]
                }
            },
            "salary": {"method": "floats_normal", "kwargs": {"mean": 50000.0, "std": 15000.0, "decimals": 2}},
            "hire_date": {
                "method": "dates",
                "kwargs": {"start": "2018-01-01", "end": "2025-10-30", "date_format": "%Y-%m-%d"}
            },
            "age": {"method": "integers", "kwargs": {"min": 22, "max": 65, "int_type": "int32"}},
            "is_remote": {"method": "booleans", "kwargs": {"true_prob": 0.30}},
            "performance_score": {"method": "floats", "kwargs": {"min": 0.0, "max": 10.0, "decimals": 1}}
        }

    @classmethod
    def sensors(cls) -> Dict[str, Any]:
        """
        Return the IoT sensor readings RandSpec (7 Fields).

        Returns
        -------
        dict
            RandSpec mapping each of the 7 fields to its method and kwargs.

        Examples
        --------
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> list(DataGenerator(RandSpecs.sensors(), seed=1).size(5).get_df().columns)
        ['sensor_id', 'device_name', 'temperature', 'humidity', 'battery_level', 'timestamp', 'is_online']
        """
        return {
            "sensor_id": {"method": "uuid4", "kwargs": {}},
            "device_name": {"method": "int_zfilled", "kwargs": {"length": 6}},
            "temperature": {"method": "floats_normal", "kwargs": {"mean": 25.0, "std": 5.0, "decimals": 1}},
            "humidity": {"method": "floats", "kwargs": {"min": 30.0, "max": 90.0, "decimals": 1}},
            "battery_level": {"method": "integers", "kwargs": {"min": 0, "max": 100, "int_type": "int32"}},
            "timestamp": {
                "method": "unix_timestamps",
                "kwargs": {"start": "2025-10-01", "end": "2025-10-30", "date_format": "%Y-%m-%d"}
            },
            "is_online": {"method": "booleans", "kwargs": {"true_prob": 0.95}}
        }

    @classmethod
    def users(cls) -> Dict[str, Any]:
        """
        Return the application users RandSpec (7 Fields).

        Returns
        -------
        dict
            RandSpec mapping each of the 7 fields to its method and kwargs.

        Examples
        --------
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> list(DataGenerator(RandSpecs.users(), seed=1).size(5).get_df().columns)
        ['user_id', 'username', 'signup_date', 'login_count', 'subscription_plan', 'is_active', 'engagement_score']
        """
        return {
            "user_id": {"method": "uuid4", "kwargs": {}},
            "username": {"method": "int_zfilled", "kwargs": {"length": 10}},
            "signup_date": {
                "method": "dates",
                "kwargs": {"start": "2023-01-01", "end": "2025-10-30", "date_format": "%Y-%m-%d"}
            },
            "login_count": {"method": "integers", "kwargs": {"min": 0, "max": 1000, "int_type": "int32"}},
            "subscription_plan": {
                "method": "distincts_prop",
                "kwargs": {"distincts": {"free": 60, "basic": 25, "premium": 10, "enterprise": 5}}
            },
            "is_active": {"method": "booleans", "kwargs": {"true_prob": 0.80}},
            "engagement_score": {"method": "floats", "kwargs": {"min": 0.0, "max": 100.0, "decimals": 1}}
        }

    @classmethod
    def events(cls) -> Dict[str, Any]:
        """
        Return the system event logs RandSpec (6 Fields).

        Returns
        -------
        dict
            RandSpec mapping each of the 6 fields to its method and kwargs.

        Examples
        --------
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> list(DataGenerator(RandSpecs.events(), seed=1).size(5).get_df().columns)
        ['event_id', 'event_type', 'timestamp', 'duration_ms', 'is_error', 'severity']
        """
        return {
            "event_id": {"method": "uuid4", "kwargs": {}},
            "event_type": {
                "method": "distincts",
                "kwargs": {
                    "distincts": ["user_login", "user_logout", "page_view",
                                "api_call", "database_query", "file_upload"]
                }
            },
            "timestamp": {
                "method": "unix_timestamps",
                "kwargs": {"start": "2025-10-01", "end": "2025-10-30", "date_format": "%Y-%m-%d"}
            },
            "duration_ms": {"method": "integers", "kwargs": {"min": 0, "max": 5000, "int_type": "int32"}},
            "is_error": {"method": "booleans", "kwargs": {"true_prob": 0.10}},
            "severity": {
                "method": "distincts_prop",
                "kwargs": {"distincts": {"info": 70, "warning": 20, "error": 8, "critical": 2}}
            }
        }

    @classmethod
    def sales(cls) -> Dict[str, Any]:
        """
        Return the sales records RandSpec (8 Fields).

        Returns
        -------
        dict
            RandSpec mapping each of the 8 fields to its method and kwargs.

        Examples
        --------
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> list(DataGenerator(RandSpecs.sales(), seed=1).size(5).get_df().columns)
        ['transaction_id', 'sale_id', 'product_id', 'amount', 'quantity', 'sales_rep', 'sale_date', 'is_wholesale']
        """
        return {
            "transaction_id": {"method": "uuid4", "kwargs": {}},
            "sale_id": {"method": "int_zfilled", "kwargs": {"length": 10}},
            "product_id": {"method": "int_zfilled", "kwargs": {"length": 8}},
            "amount": {"method": "floats_normal", "kwargs": {"mean": 500.0, "std": 200.0, "decimals": 2}},
            "quantity": {"method": "integers", "kwargs": {"min": 1, "max": 50, "int_type": "int32"}},
            "sales_rep": {
                "method": "distincts",
                "kwargs": {
                    "distincts": ["Alice Smith", "Bob Johnson", "Carol Williams",
                                "David Brown", "Emma Davis", "Frank Miller"]
                }
            },
            "sale_date": {
                "method": "dates",
                "kwargs": {"start": "2025-01-01", "end": "2025-10-30", "date_format": "%Y-%m-%d"}
            },
            "is_wholesale": {"method": "booleans", "kwargs": {"true_prob": 0.20}}
        }

    @classmethod
    def devices(cls) -> Dict[str, Any]:
        """
        Return the device inventory RandSpec (7 Fields).

        Returns
        -------
        dict
            RandSpec mapping each of the 7 fields to its method and kwargs.

        Examples
        --------
        >>> from rand_engine import DataGenerator, RandSpecs
        >>> list(DataGenerator(RandSpecs.devices(), seed=1).size(5).get_df().columns)
        ['device_id', 'serial_number', 'firmware_version', 'uptime_hours', 'status', 'is_critical', 'last_seen']
        """
        return {
            "device_id": {"method": "uuid4", "kwargs": {}},
            "serial_number": {"method": "int_zfilled", "kwargs": {"length": 12}},
            "firmware_version": {
                "method": "distincts",
                "kwargs": {"distincts": ["v1.0.0", "v1.1.0", "v1.2.0", "v2.0.0", "v2.1.0"]}
            },
            "uptime_hours": {"method": "floats_normal", "kwargs": {"mean": 500.0, "std": 200.0, "decimals": 1}},
            "status": {
                "method": "distincts_prop",
                "kwargs": {"distincts": {"online": 80, "offline": 10, "maintenance": 7, "error": 3}}
            },
            "is_critical": {"method": "booleans", "kwargs": {"true_prob": 0.15}},
            "last_seen": {
                "method": "unix_timestamps",
                "kwargs": {"start": "2025-10-01", "end": "2025-10-30", "date_format": "%Y-%m-%d"}
            }
        }