-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy path_str_builder.py
More file actions
44 lines (37 loc) · 1.1 KB
/
_str_builder.py
File metadata and controls
44 lines (37 loc) · 1.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from string import Formatter
import polars as pl
from pyochain import Dict, Iter, option
EMPTY_STR = pl.lit("")
def format_kwords(
txt: str, *, ignore_nulls: bool = False, **kwargs: pl.Expr
) -> pl.Expr:
kword_map = Dict.from_ref(kwargs)
if ignore_nulls:
kword_map = _ignore_nulls(kword_map)
return (
Iter(Formatter().parse(txt))
.map_star(lambda lit, field, _fmt, _conv: (lit, option(field)))
.collect()
.into(
lambda parts: pl.format(
parts
.iter()
.map_star(
lambda lit, field: field.map(lambda _: f"{lit}{{}}").unwrap_or(lit)
)
.join(""),
*parts
.iter()
.filter_map_star(lambda _lit, field: field)
.filter_map(kword_map.get_item),
)
)
)
def _ignore_nulls(kwargs: Dict[str, pl.Expr]) -> Dict[str, pl.Expr]:
return (
kwargs
.items()
.iter()
.map_star(lambda k, v: (k, v.fill_null(EMPTY_STR)))
.collect(Dict)
)