https://github.com/stitchfix/hamilton logo
Join Slack
Powered by
This message was deleted.
# hamilton-help
s
This message was deleted.
👀 1
s
question — what is the data source? Is it static like in your example? or?
and just so I understand, you have some data — that may or may not have various tags specified — and given a category, or subcategory, you want to get out something like as list of products and category, that match the “filter” you provided?
b
in this example, the data is static
but in reality we probably hit a database
all the data would always have all the tags specified, but the motivation for something like this is more to answer the question “What is the corresponding category to this subcategory tag?”
s
oh I see. I misunderstood the output.
want to jump on a quick call?
b
i can in about 20 minutes
👍 1
s
one option - drop the 
category_tag
and 
subcateogry_tag
functions, and instead provide them as inputs + setting defaults on the 
category
and 
subcategory
functions. E.g.
Copy code
def subcategory(data: list, subcategory_tag: str = None) -> str:
    if subcategory_tag is None:
        return None
    d = {d['subcategory_tag']: d['subcategory'] for d in data}
    return d[subcategory_tag]


def category(data: list, subcategory: str, category_tag: str = None) -> str:
    if category_tag is None and subcategory is None:
        raise ValueError

    if category_tag is not None:
        d = {d['category_tag']: d['category'] for d in data}
    else:
        d = {d['subcategory']: d['category'] for d in data}

    val = category_tag if category_tag is not None else subcategory
    return d[val]
and thus the driver looks like:
Copy code
dr = driver.Driver({}, my_functions, adapter=SimplePythonGraphAdapter(DictResult))

result_1 = dr.execute(['category'], inputs={"category_tag": "PB"})
print(result_1)

result_2 = dr.execute(['category'], inputs={"subcategory_tag": "NPB"})
print(result_2)
option 2: create two DAGs one for each case.
Copy code
def subcategory(data: list, subcategory_tag: str) -> str:
    d = {d['subcategory_tag']: d['subcategory'] for d in data}
    return d[subcategory_tag]


@config.when_not(category_tag=None)
def category__cat_tag(data: list, category_tag: str) -> str:
    d = {d['category_tag']: d['category'] for d in data}
    return d[category_tag]


@config.when_not(subcategory_tag=None)
def category__subcat_tag(data: list, subcategory: str) -> str:
    d = {d['subcategory']: d['category'] for d in data}
    return d[subcategory]
and then driver code:
Copy code
dr = driver.Driver({"category_tag": "PB"}, my_functions, adapter=SimplePythonGraphAdapter(DictResult))
result_1 = dr.execute(['category'])
print(result_1)

dr = driver.Driver({"subcategory_tag": "NPB"}, my_functions, adapter=SimplePythonGraphAdapter(DictResult))
result_2 = dr.execute(['category'])
print(result_2)
b
ok great thank you — I’m free for a call now if you are
s
yeah sorry give me 5 minutes — need to deal with my son.
b
no problem!
e
(also, as a meta-note, thank you so much for including code in your original post! Made it super easy to reason about 🙌 )
👍 1
2 Views
Open in Slack
PreviousNext
Powered by