Contract Store

With all our contracts define we can finally start using the feature_views and model_contracts. This is done through the ContractStore, which can leverage all contracts in all kinds of ways.

Load the Contracts

The first step is to register all contracts into our store. This can be done in different ways, depending on which behaviour you want.

Manually add them

The method with the least amout of magic will be to manually add them, one by one. Which can be done with the following

from aligned import ContractStore

from src.contracts import RedWine, WhiteWine, Wine, WineModel

store = ContractStore.empty()

store.add(RedWine)
store.add(WhiteWine)
store.add(Wine)
store.add(WineModel)

However, this can quickly become a pain to do. As a result can you also use the following methods.

Auto detection from directory

One simplified way of finding all contracts will be to use the .from_dir() method.

This will read all Python files in a directory, load the python module and add all feature views and model contracts.

store = await ContractStore.from_dir(".")

Auto detection from glob

Using the .from_dir() method can sometimes lead to loading python files that we do not want. Potentially leading to crashes, or very slow startup times.

As a result is it also possible to load only a subset of files based on a glob.

store = await ContractStore.from_glob("src/**/contracts.py")

From JSON file

In a scenario where the contracts needs to be shared across projects will the auto detection method not work, as the files will be missing.

As a result is it also possible to deserialize the contracts into JSON objects. Making it also possible to use something like a AWS S3 json file as the contract store.

store = await AwsS3Config(...).json_at(
    "contracts.json"
).as_contract_store()

Update Sources

Sometimes you may want to change the sources for a spesific contract. This could either be due to unit or integration testing, or because you want a source with lower latency. Such as an online source like redis, or an in memory source.

Because of this do aligned offer a few different ways of updating sources.

Dummy Store

For unit and integration testing can it sometimes be nice to mainly have random data sources. For such scenarios can you use the .dummy_store().

store = await ContractStore.from_glob("**/*.py")
test_store = store.dummy_store()

However, you often want to inject some spesific data in your tests to have some controll, and that is when the other update methods makes sense.

Update Source for Location

If there is one spesific source you want to update. Then the .update_source_for(...) method could be useful.

store = await ContractStore.from_glob("**/*.py")

new_store = store.update_source_for(
    "feature_view:titanic",
    RandomDataSource.with_values({
        "passenger_id": [1, 2, 3, 4],
        "age": [10, 0.3, 90, 65]
    })
)

Update Source of Type

Sometimes you might want to update a spesific type of source. For instance to change which secret settings to use. In those scenarios can you use the .sources_of_type(...) method.

from aligned.source.azure_blob_storage import AzureBlobConfig, AzureConfigurable

store = await ContractStore.from_glob("**/*.py")

blob_config = AzureBlobConfig(...)

store.sources_of_type(
    AzureConfigurable,
    lambda source: source.config = blob_config
)