aligned.ContractStore

class documentation

class ContractStore: (source)

Constructors: ContractStore.empty(), ContractStore.experimental(), ContractStore.from_contracts(contracts), ContractStore.from_definition(repo), ContractStore.from_dir(path), ContractStore.from_glob(glob), ContractStore.from_reference_at_path(path, reference_file), ContractStore(feature_views, models, vector_indexes, sources)

View In Hierarchy

Undocumented

Static Method empty Creates a feature store with no features or models.
Static Method experimental Undocumented
Static Method from_contracts Undocumented
Static Method from_definition Creates a feature store based on a repo definition A feature source can also be defined if wanted, otherwise will the batch source be used for reads
Async Static Method from_dir Reads and generates a feature store based on the given directory's content.
Async Static Method from_glob Reads and generates a feature store based on the given glob path.
Async Static Method from_reference_at_path Looks for a file reference struct, and loads the associated repo.
Method __init__ Undocumented
Method add Adds a feature view or a model contract
Method add_compiled_model Undocumented
Method add_compiled_view Compiles and adds the feature view to the store
Method add_feature_view Undocumented
Method add_model Compiles and adds the model to the store
Method add_view Compiles and adds the feature view to the store
Method combined_with Combines two different feature stores together.
Method dummy_store Creates a new contract store that only generates dummy data
Method event_triggers_for Undocumented
Method execute_sql Undocumented
Method feature_view Selects a feature view based on a name.
Method features_for Returns a set of features given a set of entities.
Method features_for_request Undocumented
Async Method freshness_for Undocumented
Async Method insert_into Undocumented
Method model Selects a model for easy of use.
Method model_features_for Undocumented
Method needed_configs_for Undocumented
Method needed_entities_for Undocumented
Async Method overwrite Undocumented
Method predict_when_missing Undocumented
Method raise_on_missing_config_for Undocumented
Method remove Removing a feature view or a model contract from the store.
Method repo_definition Undocumented
Method requests_for Undocumented
Method requests_for_features Undocumented
Method source_for Undocumented
Method sources_of_type Process all sources of a specific type
Method update_source_for Undocumented
Async Method upsert_into Undocumented
Method vector_index Undocumented
Method without_model_cache Undocumented
Method write_request_for Undocumented
Async Method write_to Undocumented
Instance Variable feature_views Undocumented
Instance Variable models Undocumented
Instance Variable sources Undocumented
Instance Variable vector_indexes Undocumented
Property all_models Undocumented
Property source_types Undocumented
Static Method _requests_for Undocumented
@staticmethod
def empty() -> ContractStore: (source)

Creates a feature store with no features or models.

Examples:

```python store = ContractStore.empty()

store.add(MyFeatureView) store.add(MyModel)

df = await store.execute_sql("SELECT * FROM my_view LIMIT 10").to_polars() ```

Returns:
ContractStore: An empty store with new views or models
@staticmethod
def experimental() -> ContractStore: (source)

Undocumented

@staticmethod
def from_contracts(contracts: list[FeatureViewWrapper | ModelContractWrapper]) -> ContractStore: (source)

Undocumented

@staticmethod
def from_definition(repo: RepoDefinition) -> ContractStore: (source)

Creates a feature store based on a repo definition A feature source can also be defined if wanted, otherwise will the batch source be used for reads

` repo_file: bytes = ... repo_def = RepoDefinition.from_json(repo_file) feature_store = FeatureStore.from_definition(repo_def) `

Args:
repo (RepoDefinition): The definition to setup
Returns:
FeatureStore: A ready to use feature store
@staticmethod
async def from_dir(path: str = '.') -> ContractStore: (source)

Reads and generates a feature store based on the given directory's content.

This will read the feature views, services etc in a given repo and generate a feature store. This can be used for fast development purposes.

If you rather want a more flexible deployable solution. Consider using FeatureStore.from_reference_at_path(...) which will can read an existing generated file from different storages, based on an environment variable.

Args:
path (str, optional): the directory to read from. Defaults to ".".
Returns:
FeatureStore: The generated feature store
@staticmethod
async def from_glob(glob: str) -> ContractStore: (source)

Reads and generates a feature store based on the given glob path.

This will read the feature views, services etc in a given repo and generate a feature store. This can be used for fast development purposes.

Args:
glob (str): the files to read. E.g. src/**/*.py
Returns:
ContractStore: The generated contract store
@staticmethod
async def from_reference_at_path(path: str = '.', reference_file: str = 'feature_store_location.py') -> ContractStore: (source)

Looks for a file reference struct, and loads the associated repo.

This can be used for changing which feature store definitions to read based on defined environment variables.

If you rather want to generate a feature store based on a dir, then consider using FeatureStore.from_dir(...) instead.

Args:
path (str, optional): The path of the dir to search. Defaults to ".".
Returns:
FeatureStore: A feature store based on the feature references
def __init__(self, feature_views: dict[str, CompiledFeatureView], models: dict[str, ModelSchema], vector_indexes: dict[str, ModelSchema] | None = None, sources: dict[FeatureLocation, BatchDataSource] | None = None): (source)

Undocumented

def add(self, contract: FeatureViewWrapper | ModelContractWrapper): (source)

Adds a feature view or a model contract

Args:
contract (FeatureViewWrapper | ModelContractWrappe): The contract to add
Examples:

```python @feature_view(...) class MyFeatures:

feature_id = String().as_entity() feature = Int32()

store.add(MyFeatures) ```

def add_compiled_model(self, model: ModelSchema): (source)

Undocumented

def add_compiled_view(self, view: CompiledFeatureView): (source)

Compiles and adds the feature view to the store

Examples:

```python @feature_view(...) class MyFeatureView:

id = Int32().as_entity()

my_feature = String()

store.add_compiled_view(MyFeatureView.compile()) ```

Args:
view (CompiledFeatureView): The feature view to add
def add_feature_view(self, feature_view: FeatureView | FeatureViewWrapper | CompiledFeatureView): (source)

Undocumented

def add_model(self, model: ModelContractWrapper): (source)

Compiles and adds the model to the store

Args:
model (Model): The model to add
def add_view(self, view: CompiledFeatureView | FeatureView | FeatureViewWrapper): (source)

Compiles and adds the feature view to the store

Examples:

```python @feature_view(...) class MyFeatureView:

id = Int32().as_entity()

my_feature = String()

store.add_compiled_view(MyFeatureView.compile()) ```

Args:
view (CompiledFeatureView): The feature view to add
def combined_with(self, other: ContractStore) -> ContractStore: (source)

Combines two different feature stores together.

def dummy_store(self) -> ContractStore: (source)

Creates a new contract store that only generates dummy data

def event_triggers_for(self, feature_view: str) -> set[EventTrigger]: (source)

Undocumented

def execute_sql(self, query: str) -> RetrievalJob: (source)

Undocumented

def feature_view(self, view: str | FeatureViewWrapper) -> FeatureViewStore: (source)

Selects a feature view based on a name.

From here can you query the feature view for features.

`python data = await store.feature_view('my_view').all(limit=10).to_pandas() `

Args:
view (str): The name of the feature view
Returns:
FeatureViewStore: The selected feature view ready for querying
def features_for(self, entities: ConvertableToRetrievalJob | RetrievalJob, features: list[str] | list[FeatureReference] | list[FeatureReferencable], event_timestamp_column: str | None = None, model_version_as_entity: bool | None = None) -> RetrievalJob: (source)

Returns a set of features given a set of entities.

`python entities = { "user_id": [1, 2, 3, ...] } job = store.features_for(entities, features=["user:time_since_last_login", ...]) data = await job.to_pandas() `

Args:
entities (dict[str, list] | RetrievalJob): The entities to load data for features (list[str]): A list of features to load. Use the format (<feature_view>:<feature>)
Returns:
RetrievalJob: A job that knows how to fetch the features
def features_for_request(self, request: FeatureRequest, entities: ConvertableToRetrievalJob | RetrievalJob, feature_names: set[str]) -> RetrievalJob: (source)

Undocumented

async def freshness_for(self, loc: FeatureLocation) -> datetime | None: (source)

Undocumented

async def insert_into(self, location: ConvertableToLocation, values: ConvertableToRetrievalJob | RetrievalJob): (source)

Undocumented

def model(self, model: str | ModelContractWrapper) -> ModelFeatureStore: (source)

Selects a model for easy of use.

`python entities = {"trip_id": [1, 2, 3, ...]} preds = await store.model("my_model").predict_over(entities).to_polars() `

Returns:
ModelFeatureStore: A new store that contains the selected model
def model_features_for(self, view_name: str) -> set[str]: (source)

Undocumented

def needed_configs_for(self, location: ConvertableToLocation) -> list[ConfigValue]: (source)

Undocumented

def needed_entities_for(self, features: list[FeatureReference]) -> set[Feature]: (source)

Undocumented

async def overwrite(self, location: ConvertableToLocation, values: ConvertableToRetrievalJob | RetrievalJob): (source)

Undocumented

def predict_when_missing(self) -> ContractStore: (source)

Undocumented

def raise_on_missing_config_for(self, location: ConvertableToLocation): (source)

Undocumented

def remove(self, location: ConvertableToLocation): (source)

Removing a feature view or a model contract from the store.

`python store.remove("feature_view:titanic") # or location = FeatureLocation.feature_view("titanic") store.remove(location) `

Args:
location (str | FeatureLocation): The contract to remove
def repo_definition(self) -> RepoDefinition: (source)

Undocumented

def requests_for(self, feature_request: RawStringFeatureRequest, event_timestamp_column: str | None = None, model_version_as_entity: bool | None = None) -> FeatureRequest: (source)

Undocumented

def requests_for_features(self, features: Iterable[str] | list[FeatureReference], event_timestamp_column: str | None = None) -> FeatureRequest: (source)

Undocumented

def source_for(self, location: ConvertableToLocation) -> BatchDataSource | None: (source)

Undocumented

def sources_of_type(self, source_type: type[T], function: Callable[[T, FeatureLocation], None]): (source)

Process all sources of a specific type

```python store = await ContractStore.from_dir()

def update_databricks_source(source: DatabricksSource, loc: FeatureLocation) -> None:
source.config = DatabricksConfig.serverless()
store.sources_of_type(
DatabricksSource, update_databricks_source

)

Args:
source_type: The source type you want to process function (Callable[[T, FeatureLocation], None]): The function that process the source
def update_source_for(self, location: ConvertableToLocation, source: BatchDataSource) -> ContractStore: (source)

Undocumented

async def upsert_into(self, location: ConvertableToLocation, values: ConvertableToRetrievalJob | RetrievalJob): (source)

Undocumented

def vector_index(self, name: str | ModelContractWrapper) -> VectorIndexStore: (source)

Undocumented

def without_model_cache(self) -> ContractStore: (source)

Undocumented

def write_request_for(self, location: FeatureLocation) -> RetrievalRequest: (source)

Undocumented

async def write_to(self, file: StorageFileReference): (source)

Undocumented

feature_views: dict[str, CompiledFeatureView] = (source)

Undocumented

models: dict[str, ModelSchema] = (source)

Undocumented

sources: dict[FeatureLocation, BatchDataSource] = (source)

Undocumented

vector_indexes: dict[str, ModelSchema] = (source)

Undocumented

@property
all_models: list[str] = (source)

Undocumented

@property
source_types: dict[str, type[BatchDataSource]] = (source)

Undocumented

@staticmethod
def _requests_for(feature_request: RawStringFeatureRequest, feature_views: dict[str, CompiledFeatureView], models: dict[str, ModelSchema], event_timestamp_column: str | None = None, model_version_as_entity: bool | None = None) -> FeatureRequest: (source)

Undocumented