Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save westandskif/8af54db27b4448b1e957a24da7ce4d5f to your computer and use it in GitHub Desktop.
Save westandskif/8af54db27b4448b1e957a24da7ce4d5f to your computer and use it in GitHub Desktop.
reddit-convtools-2021-07-08
from datetime import datetime
from decimal import Decimal
from convtools import conversion as c
parse_date = c.call_func(datetime.strptime, c.this(), "%Y-%m-%d").call_method(
"date"
)
parse_decimal = c.call_func(
Decimal,
c.this().call_method("replace", ",", ""),
)
YourOrmModelForTopPerformers = dict
converter = (
c.iter(
{
"id": c.item("ID"),
"full_name": c.item("Full Name"),
"dt": c.item("Transaction Date").pipe(parse_date),
"amount": c.item("Amount").pipe(parse_decimal),
}
)
.pipe(
c.group_by(c.item("full_name"), c.item("dt")).aggregate(
{
"full_name": c.item("full_name"),
"dt": c.item("dt"),
"transactions_number": c.ReduceFuncs.Count(),
"amount": c.ReduceFuncs.Sum(c.item("amount")),
"refunds": c.ReduceFuncs.Sum(
c.item("amount"), where=c.item("amount") < 0
),
}
)
)
.filter(c.item("amount") >= 1)
.iter(c.call_func(YourOrmModelForTopPerformers, c.this()))
.as_type(list)
.gen_converter(debug=True) # install black to see formatted code
)
data = converter([
{"ID": 1, "Full Name": "J S", "Transaction Date": "1970-02-01", "Amount": "-1,234.99"},
{"ID": 2, "Full Name": "J S", "Transaction Date": "1970-02-01", "Amount": "3,456.99"},
{"ID": 3, "Full Name": "J S", "Transaction Date": "1970-02-02", "Amount": "100"},
{"ID": 4, "Full Name": "N A", "Transaction Date": "1970-02-01", "Amount": "2"},
])
assert data == [
{'full_name': 'J S',
'dt': datetime.date(1970, 2, 1),
'transactions_number': 2,
'amount': Decimal('2222.00'),
'refunds': Decimal('-1234.99')},
{'full_name': 'J S',
'dt': datetime.date(1970, 2, 2),
'transactions_number': 1,
'amount': Decimal('100'),
'refunds': 0},
{'full_name': 'N A',
'dt': datetime.date(1970, 2, 1),
'transactions_number': 1,
'amount': Decimal('2'),
'refunds': 0}]
def group_by__v4(data_):
global labels_
_none = v_x0
signature_to_agg_data__v4 = defaultdict(AggData__v4)
for row__v4 in data_:
agg_data__v4 = signature_to_agg_data__v4[
(
row__v4["full_name"],
row__v4["dt"],
)
]
if row__v4["amount"] < 0:
if agg_data__v4.v2 is _none:
agg_data__v4.v2 = row__v4["amount"] or 0
else:
agg_data__v4.v2 = agg_data__v4.v2 + (row__v4["amount"] or 0)
if agg_data__v4.v0 is _none:
agg_data__v4.v0 = 1
agg_data__v4.v1 = row__v4["amount"] or 0
else:
agg_data__v4.v0 = agg_data__v4.v0 + 1
agg_data__v4.v1 = agg_data__v4.v1 + (row__v4["amount"] or 0)
result_ = (
{
"full_name": signature__v4[0],
"dt": signature__v4[1],
"transactions_number": (
0 if agg_data__v4.v0 is _none else agg_data__v4.v0
),
"amount": (0 if agg_data__v4.v1 is _none else agg_data__v4.v1),
"refunds": (0 if agg_data__v4.v2 is _none else agg_data__v4.v2),
}
for signature__v4, agg_data__v4 in signature_to_agg_data__v4.items()
)
filtered_result_ = [i_zu for i_zu in result_ if (i_zu["amount"] >= 1)]
return filtered_result_
def converter_i9(data_):
global labels_
return [
dict(i_z2)
for i_z2 in group_by__v4(
(
{
"id": i_6p["ID"],
"full_name": i_6p["Full Name"],
"dt": strptime_u5(
i_6p["Transaction Date"], "%Y-%m-%d"
).date(),
"amount": Decimal_3t(i_6p["Amount"].replace(",", "")),
}
for i_6p in data_
)
)
]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment