"""Implements classes for generating data by schema."""
import csv
import json
import pickle
import warnings
from typing import Any, Callable, ClassVar, Iterator, List, Optional, Sequence
from mimesis.exceptions import FieldError, SchemaError
from mimesis.locales import Locale
from mimesis.providers.generic import Generic
from mimesis.types import JSON, SchemaType, Seed
__all__ = ["BaseField", "Field", "Schema"]
[docs]class BaseField:
"""
BaseField is a class for generating data by the name of the method.
Instance of this object takes any string which represents the name
of any method of any supported data provider (:class:`~mimesis.Generic`)
and the ``**kwargs`` of the method.
See :class:`~mimesis.schema.BaseField.perform` for more details.
"""
class Meta:
base = True
def __init__(
self,
locale: Locale = Locale.DEFAULT,
seed: Seed = None,
providers: Optional[Sequence[Any]] = None,
) -> None:
"""Initialize field.
:param locale: Locale
:param seed: Seed for random.
"""
self._gen = Generic(locale, seed)
if providers:
self._gen.add_providers(*providers)
self._table = {} # type: ignore
def __str__(self) -> str:
return f"{self.__class__.__name__} <{self._gen.locale}>"
[docs]class Field(BaseField):
"""Greedy field.
The field whcih evaluates immediately.
Example:
>>> _ = Field()
>>> _('username')
Dogtag_1836
"""
def __call__(self, *args: Any, **kwargs: Any) -> Any:
return self.perform(*args, **kwargs)
[docs]class Schema:
"""Class which return list of filled schemas."""
_MIN_ITERATIONS_VALUE: ClassVar[int] = 1
__slots__ = ("_schema",)
def __init__(self, schema: SchemaType) -> None:
"""Initialize schema.
:param schema: A schema (must be a callable object).
"""
if schema and callable(schema):
self._schema = schema
else:
raise SchemaError()
[docs] def to_csv(self, file_path: str, iterations: int = 100, **kwargs: Any) -> None:
"""Export a schema as a CSV file.
:param file_path: File path.
:param iterations: The required number of rows.
:param kwargs: The keyword arguments for :py:class:`csv.DictWriter` class.
*New in version 5.3.0*
"""
data = self.create(iterations)
fieldnames = list(data[0])
with open(file_path, "w", newline="") as fp:
dict_writer = csv.DictWriter(fp, fieldnames, **kwargs)
dict_writer.writeheader()
dict_writer.writerows(data)
[docs] def to_json(self, file_path: str, iterations: int = 100, **kwargs: Any) -> None:
"""Export a schema as a JSON file.
:param file_path: File path.
:param iterations: The required number of rows.
:param kwargs: Extra keyword arguments for :py:func:`json.dump` class.
*New in version 5.3.0*
"""
data = self.create(iterations)
with open(file_path, "w") as fp:
json.dump(data, fp, **kwargs)
[docs] def to_pickle(self, file_path: str, iterations: int = 100, **kwargs: Any) -> None:
"""Export a schema as the pickled representation of the object to the file.
:param file_path: File path.
:param iterations: The required number of rows.
:param kwargs: Extra keyword arguments for :py:func:`pickle.dump` class.
*New in version 5.3.0*
"""
data = self.create(iterations)
with open(file_path, "wb") as fp:
pickle.dump(data, fp, **kwargs)
[docs] def create(self, iterations: int = 1) -> List[JSON]:
"""Creates a list of a fulfilled schemas.
.. note::
This method evaluates immediately, so be careful on creating
large datasets otherwise you're risking running out of memory.
If you need a lazy version of this method, see :meth:`iterator`.
:param iterations: Number of iterations.
:return: List of fulfilled schemas.
"""
if iterations < self._MIN_ITERATIONS_VALUE:
raise ValueError("The number of iterations must be greater than 0.")
return [self._schema() for _ in range(iterations)]
[docs] def loop(self) -> Iterator[JSON]:
"""Fulfills a schema **infinitely** in a lazy way.
This method can be useful when you have some dynamic
conditions in depend on which the generation must be interrupted.
If you're accepting all risks below and want to suppress
the warnings then use :py:class:`warnings.catch_warnings`
.. note::
Since data `mimesis` provides are limited, frequent calls of
this method can cause data duplication.
.. note::
Before using this method, ask yourself: **Do I really need this**?
In most cases, the answer is: Nah, :meth:`iterator` is enough.
.. warning::
Do not use this method without interrupt conditions, otherwise,
you're risking running out of memory.
.. warning::
**Never** (seriously) call `list()`, `tuple()` or any other callable which tries to
evaluate the whole lazy object on this method — infinite called infinite
for a reason.
:return: An infinite iterator with fulfilled schemas.
"""
warnings.warn(
"You're iterating over the infinite object! "
"The schema.loop() can cause a serious memory leak."
"Please, see: https://mimesis.name/en/latest/api.html#mimesis.schema.Schema.loop"
)
while True:
yield self._schema()
[docs] def iterator(self, iterations: int = 1) -> Iterator[JSON]:
"""Fulfills schema in a lazy way.
:param iterations: Number of iterations.
:return: List of fulfilled schemas.
"""
if iterations < self._MIN_ITERATIONS_VALUE:
raise ValueError("The number of iterations must be greater than 0.")
for item in range(iterations):
yield self._schema()
