Hi all. First, thanks for an awesome piece of software. It has been a pleasure to work with Hamilton. Question: how can I retrieve cached artifacts after a run, from a different python process? More context in the thread.

okay thanks for the context. let look at the code to see if what I want to do will work.

What code do you want to see? Everything that led to the generation of the cached artifacts?

sorry I’m looking at internal code

Hi @Guilherme Pereira de Freitas, if you have a DAG visualization, that could help ground the discussion! For DAG, with nodes

A -> B; A -> C; B -> C

After the first `.execute()`: • the

MetadataStore

will have an entry for

A, B, C

• the

ResultStore

will have an entry for

A, B, C

• an entry in the

MetadataStore

, must minimally store the

cache_key

and the

data_version

for each node. They may contain other metadata ◦ as you've found the

SQLiteMetadataStore

, there's currently a bug. It doesn't actually return the

cache_key

value. You can still use the

data_version

for retrieval though. I have fix planned for this! • to get the

dependencies_data_versions

, you can use

hamilton.caching.cache_key.decode_key()

. You will probably find the Internals section of the tutorial notebook informative When you don't have access to a driver

from hamilton.caching.stores.sqlite import SQLiteMetadataStore
from hamilton.caching.cache_key import decode_key

metadata_store = SQLiteMetadataStore(...)
run_ids = metadata_stores.get_run_ids()
run_id = run_ids[2]

# run_info is a list of {node_name: ..., code_version: ..., data_version: ...}
run_info = metadata_store.get_run(run_id)

Then, you can use the

data_version

(for the current node or from a dependency via the decoded

cache_key

)

from hamilton.caching.stores.file import FileResultStore

# select the right entry
node_name = "foo"
node_foo = next(n for n in run_info if n["node_name"] == node_name)

data_version = node_foo["data_version"]
result_store = FileResultStore(...)
result = result_store.get(data_version)

I'll try this in a bit, will get back here later.

ah cool — I missed the fileresultstore API.

I reread your initial question, and trimmed my reply. If your goal is only to access store results/values, you probably don't have to interact with

cache_key

. The relevant value is

data_version

, which can be retrieved from

MetadataStore.get_run()

and then used with

ResultStore.get(data_version)

If you have questions about the

cache_key

, I'm happy to help too! random: it's a side-effect and this may change, but currently values for

code_version

,

data_version

, and

cache_key

should be distinctive enough •

code_version

is a hash converted to a string (fixed length); it only includes alphanumeric characters ◦

c9e3377d6c5044944bd89eeb7073c730ee8707627c39906b4156c6411f056f00

•

data_version

uses base64 encoding (compressed, but variable length), so you'll fin distinctive trailing

==

◦

t-BDcMLikFSNdn4piUKy1mBcKPoEsnsYjUNzWg==

•

cache_key

is in 3 parts (and can be very long):

f"{node_name}-{code_version}-{encoding(dependencies_data_versions)}

◦

raw_data-9d727859b9fd883247c3379d4d25a35af4a56df9d9fde20c75c6375dde631c68-eF6zycvPS7UDAAeiAis=

The result store worked.

My pleasure! Caching is a newer feature that we're excited to expand upon, so let us know if you hit other edge cases. Happy to learn more about your use case too!

By the way, if anyone is interested in seeing how I'm creating pytest fixtures automatically from hamilton, happy to show that.

contribution and blog post please! 🙂

The heart of the code is this fixture (I'm setting the driver

ham_driver

outside of that reading in all the modules with dag nodes) :

@pytest.fixture()
def hamilton_fixtures(request, inputs):
    """Returns a dict of hamilton DAG fixtures.
    """
    # load the fixtures
    hamilton_vars = set(node.name for node in ham_driver.list_available_variables())
    requested_fixtures = set(request.fixturenames)
    hamilton_fixtures = hamilton_vars.intersection(requested_fixtures)

    <http://logger.info|logger.info>("Loading hamilton_fixtures...")

    # fixtures is a dictionary
    fixtures = ham_driver.execute(final_vars=hamilton_fixtures, inputs=inputs)

    logger.debug(f"Non-NULL hamilton fixtures: {[k for k in fixtures if fixtures[k] is not None]}")
    logger.debug(fixtures)
    <http://logger.info|logger.info>("[DONE] Loading hamilton_fixtures.")

    return fixtures

And then in

conftest.py

this_module = sys.modules[__name__]
for node in ham_driver.list_available_variables():
    def fixture_wrapper(varname=node.name):
        @pytest.fixture
        def hamfix(hamilton_fixtures):
            return hamilton_fixtures.get(varname)

        return hamfix

    setattr(this_module, node.name, fixture_wrapper(node.name))

So then you can just go write a

test_func(node_A, node_B)

and jump straight into the asserts for

node_A

and

node_B

.

@Guilherme Pereira de Freitas we have office hours most Tuesdays at 9:30am PT if you can make that — can chat there.

Yeah, I can make Tuesday 9:30pm. See you guys soon.

@Guilherme Pereira de Freitas it’s 9:30*am* (just confirming pointing that out)