ralsei.wrappers¶

Module Contents¶

Functions¶

Aliases¶

API¶

class ralsei.wrappers.OneToOne¶

type OneToOne = collections.abc.Callable[..., dict[str, Any]]

Row to row mapping function

def example(html: str):
    return {"name": get_name(html)}

class ralsei.wrappers.OneToMany¶

type OneToMany = collections.abc.Callable[..., collections.abc.Iterator[dict[str, Any]]]

One row to many rows mapping function

def example(html: str):
    for name in get_names(html):
        yield {"name": name}

ralsei.wrappers.into_many(fn: ralsei.wrappers.OneToOne) → ralsei.wrappers.OneToMany¶

Turn OneToOne mapping function into OneToMany

ralsei.wrappers.into_one(fn: ralsei.wrappers.OneToMany) → ralsei.wrappers.OneToOne¶

Turn OneToMany mapping function into OneToOne

Would throw an error if the input function yields more than one row

ralsei.wrappers.pop_id_fields(*id_fields: str, keep: bool = False) → collections.abc.Callable[[ralsei.wrappers.OneToMany], ralsei.wrappers.OneToMany]¶

Create function wrapper that ‘pops’ id_fields off the keyword arguments, calls the inner function without them, then re-inserts them into the output rows

>>> @pop_id_fields("id")
... def foo(a: int):
...     yield {"b": a * 2}
...
>>> next(foo(id=5, a=3))
{"id": 5, "b": 6}

Additionally, popped field names are saved into the function’s metadata, so that tasks can use them for inferring IdColumns

Parameters:¶

*id_fields: str¶

keyword arguments to pop

keep: bool = False¶

if True, the popped arguments would still be passed to the inner function

>>> @pop_id_fields("year")
... def foo(year: int, name: str):
...     yield {"html": download(year, name) }
...
>>> next(foo(year=2015, name="Tokyo"))
{"year": 2015, "json": {...}}

ralsei.wrappers.rename_input(**mapping: str) → collections.abc.Callable[[ralsei.wrappers.OneToMany], ralsei.wrappers.OneToMany]¶

Create function wrapper that remaps keyword argument names

@rename_input(a="b")
def foo(b: int):
    yield {...}

foo(a=10)

ralsei.wrappers.rename_output(**mapping: str) → collections.abc.Callable[[ralsei.wrappers.OneToMany], ralsei.wrappers.OneToMany]¶

Create function wrapper that remaps fields in the output dictionary

>>> @rename_output(a="b")
... def foo():
...     yield {"a": 5}
...
>>> next(foo())
{"b": 5}

ralsei.wrappers.add_to_input(**add_values: Any) → collections.abc.Callable[[ralsei.wrappers.OneToMany], ralsei.wrappers.OneToMany]¶

Create function wrapper that adds to the keyword arguments

@add_to_input(b="meow")
def foo(a: int, b: str):
    yield {...}

foo(a=5)

ralsei.wrappers.add_to_output(**add_values: Any) → collections.abc.Callable[[ralsei.wrappers.OneToMany], ralsei.wrappers.OneToMany]¶

Create function wrapper that adds entries to the output dictionary

>>> @add_to_output(b="meow")
... def foo():
...     yield {"a": 10}
...
>>> next(foo())
{"a": 10, "b": "meow"}

ralsei.wrappers.compose(fn: ralsei.wrappers.OneToMany, *decorators: collections.abc.Callable[[ralsei.wrappers.OneToMany], ralsei.wrappers.OneToMany]) → ralsei.wrappers.OneToMany¶

Compose multiple decorators together on a OneToMany

Parameters:¶

fn: ralsei.wrappers.OneToMany¶

base function

*decorators: collections.abc.Callable[[ralsei.wrappers.OneToMany], ralsei.wrappers.OneToMany]¶

decorators to apply

ralsei.wrappers.compose_one(fn: ralsei.wrappers.OneToOne, *decorators: collections.abc.Callable[[ralsei.wrappers.OneToMany], ralsei.wrappers.OneToMany]) → ralsei.wrappers.OneToOne¶

Compose multiple decorators together on a OneToOne

Parameters:¶

fn: ralsei.wrappers.OneToOne¶

base function

*decorators: collections.abc.Callable[[ralsei.wrappers.OneToMany], ralsei.wrappers.OneToMany]¶

decorators to apply

ralsei.wrappers.get_popped_fields(fn: collections.abc.Callable) → list[str] | None¶

Get fields popped by pop_id_fields() from the function metadata