> ## Documentation Index
> Fetch the complete documentation index at: https://docs.wyvern.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# SDK Reference
> The Wyvern SDK Reference
# wyvern
# wyvern.clients
# wyvern.clients.snowflake
#### generate\_snowflake\_ctx
```python theme={null}
def generate_snowflake_ctx() -> snowflake.connector.SnowflakeConnection
```
Generate a Snowflake context from the settings
# wyvern.service
## WyvernService Objects
```python theme={null}
class WyvernService()
```
The class to define, generate and run a Wyvern service
**Attributes**:
* `host` - The host to run the service on. Defaults to localhost.
* `port` - The port to run the service on. Defaults to 5000.
#### register\_routes
```python theme={null}
async def register_routes(
route_components: List[Type[APIRouteComponent]]) -> None
```
Register the routes for the Wyvern service
**Arguments**:
* `route_components` - The list of route components to register
**Returns**:
None
#### generate
```python theme={null}
@staticmethod
def generate(*,
route_components: Optional[List[Type[APIRouteComponent]]] = None,
realtime_feature_components: Optional[List[
Type[RealtimeFeatureComponent]]] = None,
host: str = "127.0.0.1",
port: int = 5000) -> WyvernService
```
Generate a Wyvern service
**Arguments**:
* `route_components` - The list of route components to register. Defaults to None.
* `realtime_feature_components` - The list of realtime feature components to register. Defaults to None.
* `host` - The host to run the service on. Defaults to localhost.
* `port` - The port to run the service on. Defaults to 5000.
**Returns**:
* `WyvernService` - The generated Wyvern service
#### run
```python theme={null}
@staticmethod
def run(*,
route_components: List[Type[APIRouteComponent]],
realtime_feature_components: Optional[List[
Type[RealtimeFeatureComponent]]] = None,
host: str = "127.0.0.1",
port: int = 5000)
```
Generate and run a Wyvern service
**Arguments**:
* `route_components` - The list of route components to register
* `realtime_feature_components` - The list of realtime feature components to register. Defaults to None.
* `host` - The host to run the service on. Defaults to localhost.
* `port` - The port to run the service on. Defaults to 5000.
**Returns**:
None
#### generate\_app
```python theme={null}
@staticmethod
def generate_app(*,
route_components: Optional[List[
Type[APIRouteComponent]]] = None,
realtime_feature_components: Optional[List[
Type[RealtimeFeatureComponent]]] = None,
host: str = "127.0.0.1",
port: int = 5000) -> FastAPI
```
Generate a Wyvern service and return the FastAPI app
**Arguments**:
* `route_components` - The list of route components to register. Defaults to None.
* `realtime_feature_components` - The list of realtime feature components to register. Defaults to None.
* `host` *str, optional* - The host to run the service on. Defaults to localhost.
* `port` *int, optional* - The port to run the service on. Defaults to 5000.
**Returns**:
* `FastAPI` - The generated FastAPI app
# wyvern.config
## Settings Objects
```python theme={null}
class Settings(BaseSettings)
```
Settings for the Wyvern service
Extends from BaseSettings class, allowing values to be overridden by environment variables. This is useful
in production for secrets you do not wish to save in code
**Attributes**:
* `ENVIRONMENT` - The environment the service is running in. Default to `development`.
* `PROJECT_NAME` - The name of the project. Default to `default`.
* `REDIS_HOST` - The host of the redis instance. Default to `localhost`.
* `REDIS_PORT` - The port of the redis instance. Default to `6379`.
* `WYVERN_API_KEY` - The API key for the Wyvern API. Default to `""`, empty string.
* `WYVERN_BASE_URL` - The base url of the Wyvern API. Default to `https://api.wyvern.ai`
WYVERN\_ONLINE\_FEATURES\_PATH:
The path to the online features endpoint. Default to `/feature/get-online-features`.
WYVERN\_HISTORICAL\_FEATURES\_PATH:
The path to the historical features endpoint. Default to `/feature/get-historical-features`.
* `WYVERN_FEATURE_STORE_URL` - The url of the Wyvern feature store. Default to `https://api.wyvern.ai`.
* `SNOWFLAKE_ACCOUNT` - The account name of the Snowflake instance. Default to `""`, empty string.
* `SNOWFLAKE_USER` - The username of the Snowflake instance. Default to `""`, empty string.
* `SNOWFLAKE_PASSWORD` - The password of the Snowflake instance. Default to `""`, empty string.
* `SNOWFLAKE_ROLE` - The role of the Snowflake instance. Default to `""`, empty string.
* `SNOWFLAKE_WAREHOUSE` - The warehouse of the Snowflake instance. Default to `""`, empty string.
* `SNOWFLAKE_DATABASE` - The database of the Snowflake instance. Default to `""`, empty string.
* `SNOWFLAKE_OFFLINE_STORE_SCHEMA` - The schema of the Snowflake instance. Default to `PUBLIC`.
* `AWS_ACCESS_KEY_ID` - The access key id for the AWS instance. Default to `""`, empty string.
* `AWS_SECRET_ACCESS_KEY` - The secret access key for the AWS instance. Default to `""`, empty string.
* `AWS_REGION_NAME` - The region name for the AWS instance. Default to `us-east-1`.
* `FEATURE_STORE_TIMEOUT` - The timeout for the feature store. Default to `60` seconds.
* `SERVER_TIMEOUT` - The timeout for the server. Default to `60` seconds.
* `REDIS_BATCH_SIZE` - The batch size for the redis instance. Default to `100`.
* `WYVERN_INDEX_VERSION` - The version of the Wyvern index. Default to `1`.
* `MODELBIT_BATCH_SIZE` - The batch size for the modelbit. Default to `30`.
* `EXPERIMENTATION_ENABLED` - Whether experimentation is enabled. Default to `False`.
* `EXPERIMENTATION_PROVIDER` - The experimentation provider. Default to `ExperimentationProvider.EPPO.value`.
* `EPPO_API_KEY` - The API key for EPPO (an experimentation provider). Default to `""`, empty string.
* `FEATURE_STORE_ENABLED` - Whether the feature store is enabled. Default to `True`.
* `EVENT_LOGGING_ENABLED` - Whether event logging is enabled. Default to `True`.
# wyvern.core.compression
#### wyvern\_encode
```python theme={null}
def wyvern_encode(data: Dict[str, Any]) -> bytes
```
encode a dict to compressed bytes using lz4.frame
#### wyvern\_decode
```python theme={null}
def wyvern_decode(data: Union[bytes, str]) -> Dict[str, Any]
```
decode compressed bytes to a dict with lz4.frame
# wyvern.core
# wyvern.core.http
## AiohttpClientWrapper Objects
```python theme={null}
class AiohttpClientWrapper()
```
AiohttpClientWrapper is a singleton wrapper around aiohttp.ClientSession.
#### start
```python theme={null}
def start()
```
Instantiate the client. Call from the FastAPI startup hook.
#### stop
```python theme={null}
async def stop()
```
Gracefully shutdown. Call from FastAPI shutdown hook.
#### \_\_call\_\_
```python theme={null}
def __call__()
```
Calling the instantiated AiohttpClientWrapper returns the wrapped singleton.
#### aiohttp\_client
The aiohttp client singleton. Use this to make requests.
**Example**:
```python theme={null}
from wyvern.core.http import aiohttp_client
aiohttp_client().get("https://www.wyvern.ai")
```
# wyvern.experimentation.experimentation\_logging
## ExperimentationEventData Objects
```python theme={null}
class ExperimentationEventData(BaseModel)
```
Data class for ExperimentationEvent.
**Attributes**:
* `experiment_id` - The experiment id.
* `entity_id` - The entity id.
* `result` - The result of the experiment. Can be None.
* `timestamp` - The timestamp of the event.
* `metadata` - The metadata of the event such as targeting parameters etc.
* `has_error` - Whether the request has errored or not.
## ExperimentationEvent Objects
```python theme={null}
class ExperimentationEvent(LoggedEvent[ExperimentationEventData])
```
Event class for ExperimentationEvent.
**Attributes**:
* `event_type` - The event type. This is always EventType.EXPERIMENTATION.
# wyvern.experimentation.providers
# wyvern.experimentation.providers.eppo\_provider
## EppoExperimentationClient Objects
```python theme={null}
class EppoExperimentationClient(BaseExperimentationProvider)
```
An experimentation client specifically for the Eppo platform.
Extends the BaseExperimentationProvider to provide functionality using the Eppo client.
**Methods**:
* **init**() -> None
* get\_result(experiment\_id: str, entity\_id: str, \*\*kwargs) -> str
* log\_result(experiment\_id: str, entity\_id: str, variant: str) -> None
#### get\_result
```python theme={null}
def get_result(experiment_id: str, entity_id: str, **kwargs) -> str
```
Fetches the variant for a given experiment and entity from the Eppo client.
**Arguments**:
* experiment\_id (str): The unique ID of the experiment.
* entity\_id (str): The unique ID of the entity (e.g., user or other subject).
* \*\*kwargs: Additional arguments to be passed to the Eppo client's get\_assignment method.
**Returns**:
* str: The assigned variant for the given experiment and entity.
#### log\_result
```python theme={null}
def log_result(experiment_id: str,
entity_id: str,
variant: Optional[str] = None,
has_error: bool = False,
**kwargs) -> None
```
Logs the result for a given experiment and entity.
**Arguments**:
* experiment\_id (str): The unique ID of the experiment.
* entity\_id (str): The unique ID of the entity.
* variant (str): The assigned variant for the given experiment and entity.
* `Note` - This method is overridden to do nothing because the assignment logger we set in Eppo already
handles result logging upon assignment.
# wyvern.experimentation.providers.base
## ExperimentationProvider Objects
```python theme={null}
class ExperimentationProvider(str, Enum)
```
An enum for the experimentation providers.
## BaseExperimentationProvider Objects
```python theme={null}
class BaseExperimentationProvider(ABC)
```
A base class for experimentation providers.
All providers should inherit from this and implement the necessary methods.
#### get\_result
```python theme={null}
@abstractmethod
def get_result(experiment_id: str, entity_id: str, **kwargs) -> str
```
Get the result (variant) for a given experiment and entity.
**Arguments**:
* experiment\_id (str): The unique ID of the experiment.
* entity\_id (str): The unique ID of the entity.
* kwargs (dict): Any additional arguments to pass to the provider for targeting.
**Returns**:
* str: The result (variant) assigned to the entity for the specified experiment.
#### log\_result
```python theme={null}
@abstractmethod
def log_result(experiment_id: str,
entity_id: str,
variant: Optional[str] = None,
has_error: bool = False,
**kwargs) -> None
```
Log the result (variant) for a given experiment and entity.
**Arguments**:
* experiment\_id (str): The unique ID of the experiment.
* entity\_id (str): The unique ID of the entity.
* variant (str): The result (variant) assigned to the entity for the specified experiment.
* kwargs (dict): Any additional arguments to pass to the provider for targeting.
**Returns**:
* None
# wyvern.experimentation.client
## ExperimentationClient Objects
```python theme={null}
class ExperimentationClient()
```
A client for interacting with experimentation providers.
#### \_\_init\_\_
```python theme={null}
def __init__(provider_name: str, api_key: Optional[str] = None)
```
Initializes the ExperimentationClient with a specified provider.
**Arguments**:
* provider\_name (str): The name of the experimentation provider (e.g., "eppo").
#### get\_experiment\_result
```python theme={null}
def get_experiment_result(experiment_id: str, entity_id: str,
**kwargs) -> Optional[str]
```
Get the result (variant) for a given experiment and entity using the chosen provider.
**Arguments**:
* experiment\_id (str): The unique ID of the experiment.
* entity\_id (str): The unique ID of the entity.
* kwargs (dict): Any additional arguments to pass to the provider for targeting.
**Returns**:
* str: The result (variant) assigned to the entity for the specified experiment.
# wyvern.experimentation
# wyvern.index
# wyvern.request\_context
#### current
```python theme={null}
def current() -> Optional[WyvernRequest]
```
Get the current request context
**Returns**:
The current request context, or None if there is none
#### ensure\_current\_request
```python theme={null}
def ensure_current_request() -> WyvernRequest
```
Get the current request context, or raise an error if there is none
**Returns**:
The current request context if there is one
**Raises**:
* `RuntimeError` - If there is no current request context
#### set
```python theme={null}
def set(request: WyvernRequest) -> None
```
Set the current request context
**Arguments**:
* `request` - The request context to set
**Returns**:
None
#### reset
```python theme={null}
def reset() -> None
```
Reset the current request context
**Returns**:
None
# wyvern.wyvern\_typing
#### WyvernFeature
A WyvernFeature defines the type of a feature in Wyvern. It can be a float, a string, a list of floats, or None.
# wyvern.cli.commands
#### init
```python theme={null}
@app.command()
def init(project: str = typer.Argument(...,
help="Name of the project")) -> None
```
Initializes Wyvern application template code
**Arguments**:
* `project` *str* - Name of the project
#### run
```python theme={null}
@app.command()
def run(
path: str = "pipelines.main:app",
host: Annotated[
str,
typer.Option(help="Host to run the application on"),
] = "0.0.0.0",
port: Annotated[
int,
typer.Option(
help="Port to run the application on. Default port is 5001"),
] = 5001
) -> None
```
Starts Wyvern application server
Example usage:
wyvern run --path pipelines.main:app --host 0.0.0.0 --port 5001
**Arguments**:
* `path` *str* - path to the wyvern app. Default path is pipelines.main:app
* `host` *str* - Host to run the application on. Default host is 0.0.0.0
* `port` *int* - Port to run the application on. Default port is 5001
#### redis
```python theme={null}
@app.command()
def redis() -> None
```
Starts Redis server. This command will also install redis locally if it's not installed.
# wyvern.api
#### ensure\_async\_client
```python theme={null}
def ensure_async_client(func: Callable) -> Callable
```
Ensure that the async client is open before calling the function and close it after calling the function
**Arguments**:
* `func` - The function to be wrapped
**Returns**:
The wrapped function
## WyvernAPI Objects
```python theme={null}
class WyvernAPI()
```
Wyvern API client
#### get\_historical\_features
```python theme={null}
@ensure_async_client
def get_historical_features(
features: List[str], entities: Union[Dict[Hashable, List[Any]],
pd.DataFrame]) -> pd.DataFrame
```
Aggregate all the historical features, including the offline features in your data warehouse
and the historical real-time features being consumed by wyvern pipeline.
**Arguments**:
* `features` - A list of feature names.
* `entities` - A dictionary or pandas DataFrame of entity names and their values.
some requirements of entities:
* entities must have request and timestamp keys
* request is a list of the request\_id of request getting into Wyvern's pipeline
* timestamp is a list of timestamp of the request
* the rest of the columns are the entity for the features and the user interaction data
**Returns**:
A pandas DataFrame with all the feature data you're requesting from the entities.
# wyvern.components.candidates
# wyvern.components.candidates.candidate\_logger
## CandidateEventData Objects
```python theme={null}
class CandidateEventData(EntityEventData)
```
Event data for a candidate event
**Attributes**:
* `candidate_score` - The score of the candidate
* `candidate_order` - The order of the candidate in the list of candidates
# wyvern.components.pagination.pagination\_component
## PaginationRequest Objects
```python theme={null}
class PaginationRequest(GenericModel, Generic[T])
```
This is the input to the PaginationComponent.
**Attributes**:
* `pagination_fields` - The pagination fields that are used to compute the pagination.
* `entities` - The entities that need to be paginated.
## PaginationComponent Objects
```python theme={null}
class PaginationComponent(Component[PaginationRequest[T], List[T]])
```
This component is used to paginate the entities. It takes in the pagination fields and the entities and returns
the paginated entities.
#### execute
```python theme={null}
async def execute(input: PaginationRequest[T], **kwargs) -> List[T]
```
This method paginates the entities based on the pagination fields.
Validations:
1. The ranking page should be greater than or equal to 0.
2. The candidate page should be greater than or equal to 0.
3. The candidate page size should be less than or equal to 1000.
4. The number of entities should be less than or equal to 1000.
5. The user page size should be less than or equal to 100.
6. The user page size should be less than or equal to the candidate page size.
7. The end index should be less than the number of entities.
8. The end index should be greater than the start index.
**Returns**:
The paginated entities.
# wyvern.components.pagination
# wyvern.components.pagination.pagination\_fields
## PaginationFields Objects
```python theme={null}
class PaginationFields(BaseModel)
```
Pagination fields for requests. This is a mixin class that can be used in any request that requires pagination.
**Attributes**:
* `user_page_size` - Zero-indexed user facing page number
* `user_page` - Number of items per user facing page
* `candidate_page_size` - This is the size of the candidate page.
* `candidate_page` - This is the zero-indexed page number for the candidate set
# wyvern.components.features.realtime\_features\_component
#### PRIMARY\_ENTITY
The primary entity is the entity that is the main entity for the feature. For example, if we are computing
the feature for a user, the primary entity would be the user.
#### SECONDARY\_ENTITY
The secondary entity is the entity that is the secondary entity for the feature. For example, if we are computing
the feature for a user and a product, the secondary entity would be the product. If we are computing the feature
for a user, the secondary entity would be None.
## RealtimeFeatureRequest Objects
```python theme={null}
class RealtimeFeatureRequest(GenericModel, Generic[REQUEST_ENTITY])
```
This is the request that is passed into the realtime feature component.
## RealtimeFeatureEntity Objects
```python theme={null}
class RealtimeFeatureEntity(GenericModel, Generic[PRIMARY_ENTITY,
SECONDARY_ENTITY])
```
This is the entity that is passed into the realtime feature component. It contains the primary entity and
the secondary entity. If the feature is only for the primary entity, the secondary entity will be None.
## RealtimeFeatureComponent Objects
```python theme={null}
class RealtimeFeatureComponent(Component[
Tuple[
RealtimeFeatureRequest[REQUEST_ENTITY],
RealtimeFeatureEntity[PRIMARY_ENTITY, SECONDARY_ENTITY],
],
Optional[FeatureData],
], Generic[PRIMARY_ENTITY, SECONDARY_ENTITY, REQUEST_ENTITY])
```
This is the base class for all realtime feature components. It contains the logic for computing the realtime
feature. The realtime feature component can be used to compute features for a single entity, two entities, or
a request. The realtime feature component can also be used to compute composite features for two entities.
The realtime feature component is a generic class that takes in the primary entity, secondary entity, and request
entity as type parameters. The primary entity is the entity that is the main entity for the feature. For example,
if we are computing the feature for a user, the primary entity would be the user. The secondary entity is the
entity that is the secondary entity for the feature. For example, if we are computing the feature for a user and
a product, the secondary entity would be the product. If we are computing the feature for a user, the secondary
entity would be None. The request entity is the request that is passed into the realtime feature component. We can
use the request entity to compute features for a request. For example, if we are computing the realtime features for
a ranking request, the request entity would be the ranking request. We can combine the primary entity, secondary
entity, and request entity to compute composite features.
**Attributes**:
* `NAME` - The name of the realtime feature component. This is used to identify the realtime feature component.
* `real_time_features` - A list of all the realtime feature components.
* `component_registry` - A dictionary that maps the name of the realtime feature component to the realtime feature
#### \_\_init\_\_
```python theme={null}
def __init__(*upstreams: Component,
output_feature_names: Optional[Set[str]] = None,
required_feature_names: Optional[Set[str]] = None,
name: Optional[str] = None)
```
**Arguments**:
* `name`: Name of the component
* `output_feature_names`: features outputted by this real-time feature
#### get\_type\_args\_simple
```python theme={null}
@classmethod
def get_type_args_simple(cls, index: int) -> Type
```
Get the type argument at the given index for the class. This is used to get the primary entity type, secondary
entity type, and request entity type.
#### get\_entity\_names
```python theme={null}
@classmethod
def get_entity_names(cls, full_feature_name: str) -> Optional[List[str]]
```
Get the entity identifier type, which will be used as sql column name
full\_feature\_name is of the form `:`
#### get\_entity\_type\_column
```python theme={null}
@classmethod
def get_entity_type_column(cls, full_feature_name: str) -> Optional[str]
```
Get the entity identifier type, which will be used as sql column name
full\_feature\_name is of the form `:`
#### can\_execute\_on
```python theme={null}
def can_execute_on(request: REQUEST_ENTITY,
primary_entity: Optional[PRIMARY_ENTITY],
secondary_entity: Optional[SECONDARY_ENTITY]) -> bool
```
Checks if the input matches the entity type, so we can execute on it
#### set\_full\_feature\_name
```python theme={null}
def set_full_feature_name(
feature_data: Optional[FeatureData]) -> Optional[FeatureData]
```
Sets the full feature name for the feature data
# wyvern.components.features.feature\_store
## FeatureStoreRetrievalRequest Objects
```python theme={null}
class FeatureStoreRetrievalRequest(BaseModel)
```
Request to retrieve features from the feature store.
**Attributes**:
* `identifiers` - List of identifiers for which features are to be retrieved.
* `feature_names` - List of feature names to be retrieved. Feature names are of the form
`:`.
## FeatureStoreRetrievalComponent Objects
```python theme={null}
class FeatureStoreRetrievalComponent(Component[FeatureStoreRetrievalRequest,
FeatureMap])
```
Component to retrieve features from the feature store. This component is responsible for fetching features from
the feature store and returning them in the form of a FeatureMap. The FeatureMap is a mapping from identifiers to
FeatureData. The FeatureData contains the identifier and a mapping from feature names to feature values. The
feature names are of the form `:`. The feature values are of type WyvernFeature
which is a union of all the possible feature types. The feature types are defined in `wyvern/wyvern_typing.py`.
The FeatureStoreRetrievalComponent is a singleton and can be accessed via `feature_store_retrieval_component`.
The FeatureStoreRetrievalComponent is configured via the following environment variables:
* WYVERN\_API\_KEY: if you're using Wyvern's feature store, this is the API key for Wyvern
* WYVERN\_FEATURE\_STORE\_URL: url to the feature store
* WYVERN\_ONLINE\_FEATURES\_PATH: url path to the feature store's online features endpoint
* FEATURE\_STORE\_ENABLED: whether the feature store is enabled or not
#### fetch\_features\_from\_feature\_store
```python theme={null}
async def fetch_features_from_feature_store(
identifiers: List[Identifier], feature_names: List[str]) -> FeatureMap
```
Fetches features from the feature store for the given identifiers and feature names.
**Arguments**:
* `identifiers` - List of identifiers for which features are to be retrieved.
* `feature_names` - List of feature names to be retrieved.
**Returns**:
FeatureMap containing the features for the given identifiers and feature names.
#### execute
```python theme={null}
@tracer.wrap(name="FeatureStoreRetrievalComponent.execute")
async def execute(input: FeatureStoreRetrievalRequest,
handle_exceptions: bool = False,
**kwargs) -> FeatureMap
```
Fetches features from the feature store for the given identifiers and feature names. This method is a wrapper
around `fetch_features_from_feature_store` which handles exceptions and returns an empty FeatureMap in case of
an exception.
# wyvern.components.features.feature\_retrieval\_pipeline
## FeatureRetrievalPipelineRequest Objects
```python theme={null}
class FeatureRetrievalPipelineRequest(GenericModel, Generic[REQUEST_ENTITY])
```
This is the input to the FeatureRetrievalPipeline component that is used to retrieve features.
**Attributes**:
* `request` - The request that is used to retrieve features. This is used to retrieve the entities and
identifiers that are needed to compute the features.
* `requested_feature_names` - The feature names that are
requested. This is used to filter out the real-time features that are calculated instead of
retrieved from the feature store. ie: `product_fv:FEATURE_PRODUCT_AMOUNT_PAID_LAST_15_DAYS`
* `feature_overrides` - This is used to override the default real-time features.
## FeatureRetrievalPipeline Objects
```python theme={null}
class FeatureRetrievalPipeline(
Component[FeatureRetrievalPipelineRequest[REQUEST_ENTITY],
FeatureMap], Generic[REQUEST_ENTITY])
```
This component is used to retrieve features for a given request. It is composed of the following components: 1. FeatureStoreRetrievalComponent: This component is used to retrieve features from the feature store. 2. RealtimeFeatureComponent: This component is used to compute real-time features. 3. FeatureEventLoggingComponent: This component is used to log feature events.
#### \_\_init\_\_
```python theme={null}
def __init__(*upstreams: Component,
name: str,
handle_exceptions: bool = False)
```
**Arguments**:
* `*upstreams` - The upstream components to this component.
* `name` - The name of this component.
* `handle_exceptions` - Whether to handle feature store exceptions. Defaults to False.
If True, missing feature values will be None instead of raising exceptions.
If False, exceptions will be raised.
#### execute
```python theme={null}
@tracer.wrap(name="FeatureRetrievalPipeline.execute")
async def execute(input: FeatureRetrievalPipelineRequest[REQUEST_ENTITY],
**kwargs) -> FeatureMap
```
This method is used to retrieve features for a given request.
It is composed of the following steps: 0. Figure out which features are real-time features and which features are feature store features. 1. Retrieve features from the feature store. 2. Compute real-time features. 3. Combine the feature store features and real-time features into one FeatureMap. 4. Log the feature values to the feature event logging component.
# wyvern.components.features
# wyvern.components.features.feature\_logger
## FeatureLogEventData Objects
```python theme={null}
class FeatureLogEventData(BaseModel)
```
Data for a feature event.
**Attributes**:
* `feature_identifier` - The identifier of the feature.
* `feature_identifier_type` - The type of the feature identifier.
* `feature_name` - The name of the feature.
* `feature_value` - The value of the feature.
## FeatureEvent Objects
```python theme={null}
class FeatureEvent(LoggedEvent[FeatureLogEventData])
```
A feature event.
**Attributes**:
* `event_type` - The type of the event. Defaults to EventType.FEATURE.
## FeatureEventLoggingRequest Objects
```python theme={null}
class FeatureEventLoggingRequest(GenericModel, Generic[REQUEST_ENTITY])
```
A request to log feature events.
**Attributes**:
* `request` - The request to log feature events for.
* `feature_map` - The feature map to log.
## FeatureEventLoggingComponent Objects
```python theme={null}
class FeatureEventLoggingComponent(
Component[FeatureEventLoggingRequest[REQUEST_ENTITY],
None], Generic[REQUEST_ENTITY])
```
A component that logs feature events.
#### execute
```python theme={null}
async def execute(input: FeatureEventLoggingRequest[REQUEST_ENTITY],
**kwargs) -> None
```
Logs feature events.
# wyvern.components
# wyvern.components.models.model\_component
#### MODEL\_OUTPUT\_DATA\_TYPE
MODEL\_OUTPUT\_DATA\_TYPE is the type of the output of the model. It can be a float, a string, or a list of floats
(e.g. a list of probabilities, embeddings, etc.)
## ModelEventData Objects
```python theme={null}
class ModelEventData(BaseModel)
```
This class defines the data that will be logged for each model event.
**Arguments**:
* `model_name` - The name of the model
* `model_output` - The output of the model
* `entity_identifier` - The identifier of the entity that was used to generate the model output. This is optional.
* `entity_identifier_type` - The type of the identifier of the entity that was used to generate the model output.
This is optional.
## ModelEvent Objects
```python theme={null}
class ModelEvent(LoggedEvent[ModelEventData])
```
Model event. This is the event that is logged when a model is evaluated.
**Arguments**:
* `event_type` - The type of the event. This is always EventType.MODEL.
## ModelOutput Objects
```python theme={null}
class ModelOutput(GenericModel, Generic[MODEL_OUTPUT_DATA_TYPE])
```
This class defines the output of a model.
**Arguments**:
* `data` - A dictionary mapping entity identifiers to model outputs. The model outputs can also be None.
* `model_name` - The name of the model. This is optional.
#### get\_entity\_output
```python theme={null}
def get_entity_output(
identifier: Identifier) -> Optional[MODEL_OUTPUT_DATA_TYPE]
```
Get the model output for a given entity identifier.
**Arguments**:
* `identifier` - The identifier of the entity.
**Returns**:
The model output for the given entity identifier. This can also be None if the model output is None.
## ModelInput Objects
```python theme={null}
class ModelInput(GenericModel, Generic[GENERALIZED_WYVERN_ENTITY,
REQUEST_ENTITY])
```
This class defines the input to a model.
**Arguments**:
* `request` - The request that will be used to generate the model input.
* `entities` - A list of entities that will be used to generate the model input.
#### first\_entity
```python theme={null}
@property
def first_entity() -> GENERALIZED_WYVERN_ENTITY
```
Get the first entity in the list of entities. This is useful when you know that there is only one entity.
**Returns**:
The first entity in the list of entities.
#### first\_identifier
```python theme={null}
@property
def first_identifier() -> Identifier
```
Get the identifier of the first entity in the list of entities. This is useful when you know that there is only
one entity.
**Returns**:
The identifier of the first entity in the list of entities.
## ModelComponent Objects
```python theme={null}
class ModelComponent(Component[
MODEL_INPUT,
MODEL_OUTPUT,
])
```
This class defines a model component. A model component is a component that takes in a request and a list of
entities and outputs a model output. The model output is a dictionary mapping entity identifiers to model outputs.
The model outputs can also be None if the model output is None for a given entity.
#### get\_type\_args\_simple
```python theme={null}
@classmethod
def get_type_args_simple(cls, index: int) -> Type
```
Get the type argument at the given index. This is used to get the model input and model output types.
#### manifest\_feature\_names
```python theme={null}
@cached_property
def manifest_feature_names() -> Set[str]
```
This function defines which features are necessary for model evaluation
Our system will automatically fetch the required features from the feature store
to make this model evaluation possible
#### execute
```python theme={null}
async def execute(input: MODEL_INPUT, **kwargs) -> MODEL_OUTPUT
```
The model\_name and model\_score will be automatically logged
#### batch\_inference
```python theme={null}
async def batch_inference(
request: BaseWyvernRequest, entities: List[Union[WyvernEntity,
BaseWyvernRequest]]
) -> Sequence[Optional[Union[float, str, List[float]]]]
```
Define your model inference in a batched manner so that it's easier to boost inference speed
#### inference
```python theme={null}
async def inference(input: MODEL_INPUT, **kwargs) -> MODEL_OUTPUT
```
The inference function is the main entrance to model evaluation.
By default, the base ModelComponent slices entities into smaller batches and call batch\_inference on each batch.
The default batch size is 30. You should be able to configure the MODEL\_BATCH\_SIZE env variable
to change the batch size.
In order to set up model inference, you only need to define a class that inherits ModelComponent and
implement batch\_inference.
You can also override this function if you want to customize the inference logic.
# wyvern.components.models
# wyvern.components.models.modelbit\_component
## ModelbitComponent Objects
```python theme={null}
class ModelbitComponent(ModelComponent[MODEL_INPUT, MODEL_OUTPUT])
```
ModelbitComponent is a base class for all modelbit model components. It provides a common interface to implement
all modelbit models.
ModelbitComponent is a subclass of ModelComponent.
**Attributes**:
* `AUTH_TOKEN` - A class variable that stores the auth token for Modelbit.
* `URL` - A class variable that stores the url for Modelbit.
#### \_\_init\_\_
```python theme={null}
def __init__(*upstreams,
name: Optional[str] = None,
auth_token: Optional[str] = None,
url: Optional[str] = None) -> None
```
**Arguments**:
* `*upstreams` - A list of upstream components.
* `name` - A string that represents the name of the model.
* `auth_token` - A string that represents the auth token for Modelbit.
* `url` - A string that represents the url for Modelbit.
**Raises**:
* `WyvernModelbitTokenMissingError` - If the auth token is not provided.
#### modelbit\_features
```python theme={null}
@cached_property
def modelbit_features() -> List[str]
```
This is a cached property that returns a list of modelbit features. This method should be implemented by the
subclass.
#### manifest\_feature\_names
```python theme={null}
@cached_property
def manifest_feature_names() -> Set[str]
```
This is a cached property that returns a set of manifest feature names. This method wraps around the
modelbit\_features property.
#### build\_requests
```python theme={null}
async def build_requests(
input: MODEL_INPUT) -> Tuple[List[Identifier], List[Any]]
```
Please refer to modlebit batch inference API:
[https://doc.modelbit.com/deployments/rest-api/](https://doc.modelbit.com/deployments/rest-api/)
#### inference
```python theme={null}
async def inference(input: MODEL_INPUT, **kwargs) -> MODEL_OUTPUT
```
This method sends a request to Modelbit and returns the output.
# wyvern.components.api\_route\_component
## APIRouteComponent Objects
```python theme={null}
class APIRouteComponent(Component[REQUEST_SCHEMA, RESPONSE_SCHEMA])
```
APIRouteComponent is the base class for all the API routes in Wyvern. It is a Component that
takes in a request schema and a response schema, and it is responsible for hydrating the request
data with Wyvern Index data, and then pass the hydrated data to the next component in the pipeline.
The APIRouteComponent is also responsible for the API routing, which means it is responsible for
the API versioning and the API path.
**Attributes**:
* `API_VERSION` - the version of the API. This is used in the API routing. The default value is "v1".
* `PATH` - the path of the API. This is used in the API routing.
* `REQUEST_SCHEMA_CLASS` - the class of the request schema. This is used to validate the request data.
* `RESPONSE_SCHEMA_CLASS` - the class of the response schema. This is used to validate the response data.
* `API_NAME` - the name of the API. This is used in the API routing. If not provided, the name of the
APIRouteComponent will be used.
#### warm\_up
```python theme={null}
async def warm_up(input: REQUEST_SCHEMA) -> None
```
This is the warm-up function that is called before the API route is called.
#### hydrate
```python theme={null}
@tracer.wrap(name="APIRouteComponent.hydrate")
async def hydrate(input: REQUEST_SCHEMA) -> None
```
Wyvern APIRouteComponent recursively hydrate the request input data with Wyvern Index data
TODO: this function could be moved to a global place
# wyvern.components.business\_logic.boosting\_business\_logic
## BoostingBusinessLogicComponent Objects
```python theme={null}
class BoostingBusinessLogicComponent(
BusinessLogicComponent[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY],
Generic[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY])
```
A component that performs boosting on an entity with a set of candidates. The boosting can be multiplicative
or additive.
The request itself could contain more than just entities, for example it may contain a query and so on
#### boost
```python theme={null}
def boost(
scored_candidates: List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]],
entity_keys: Set[str],
boost: float,
entity_key_mapping: Callable[
[GENERALIZED_WYVERN_ENTITY],
str,
] = lambda candidate: candidate.identifier.identifier,
multiplicative=False
) -> List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]]
```
Boosts the score of each candidate by a certain factor
**Arguments**:
* `scored_candidates` - The list of scored candidates
* `entity_keys` - The set of entity keys (unique identifiers) to boost
* `boost` - The boost factor
* `entity_key_mapping` - A lambda function that takes in a candidate entity and
returns the field we should apply the boost to
* `multiplicative` - Whether to apply the boost with multiplication or addition - true indicates it is
multiplication and false indicates it is addition
**Returns**:
The list of scored candidates with the boost applied
## CSVBoostingBusinessLogicComponent Objects
```python theme={null}
class CSVBoostingBusinessLogicComponent(
BoostingBusinessLogicComponent[GENERALIZED_WYVERN_ENTITY,
REQUEST_ENTITY],
Generic[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY])
```
This component reads a csv file and applies the boost based on specific column name, entity key, and score
combinations
Methods to define: Given a CSV row, generate the entity key and boost value
**Arguments**:
* `csv_file` - The path to the CSV file
* `multiplicative` - Whether to apply the boost with multiplication or addition - true indicates it is
multiplication and false indicates it is addition
#### initialize
```python theme={null}
async def initialize() -> None
```
Reads the CSV file and populates the lookup table
#### extract\_keys\_from\_csv\_row
```python theme={null}
@abstractmethod
async def extract_keys_from_csv_row(row: Series) -> str
```
Given a CSV row, generate the unique combinations that would apply a boost
Example, in a file that has the following:
product\_id, query, boost
The method would return a unique concatenation (ie product\_id:query)
#### extract\_boost\_value\_from\_csv\_row
```python theme={null}
@abstractmethod
async def extract_boost_value_from_csv_row(row: Series) -> float
```
Given a CSV row, generate the unique combinations that would apply a boost
Example, in a file that has the following:
product\_id, query, boost
The method would return the boost value
#### extract\_key\_from\_request\_entity
```python theme={null}
@abstractmethod
async def extract_key_from_request_entity(candidate: GENERALIZED_WYVERN_ENTITY,
request: REQUEST_ENTITY) -> str
```
Given a candidate and a request, generate a unique key that would apply a boost
#### execute
```python theme={null}
async def execute(
input: BusinessLogicRequest[
GENERALIZED_WYVERN_ENTITY,
REQUEST_ENTITY,
], **kwargs) -> List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]]
```
Boosts the score of each candidate by a certain factor
# wyvern.components.business\_logic
# wyvern.components.business\_logic.business\_logic
## BusinessLogicEventData Objects
```python theme={null}
class BusinessLogicEventData(EntityEventData)
```
The data associated with a business logic event
**Arguments**:
* `business_logic_pipeline_order` - The order of the business logic pipeline that this event occurred in
* `business_logic_name` - The name of the business logic component that this event occurred in
* `old_score` - The old score of the entity
* `new_score` - The new score of the entity
## BusinessLogicEvent Objects
```python theme={null}
class BusinessLogicEvent(LoggedEvent[BusinessLogicEventData])
```
An event that occurs in the business logic layer
## BusinessLogicRequest Objects
```python theme={null}
class BusinessLogicRequest(GenericModel, Generic[GENERALIZED_WYVERN_ENTITY,
REQUEST_ENTITY])
```
A request to the business logic layer to perform business logic on a set of candidates
**Arguments**:
* `request` - The request that the business logic layer is being asked to perform business logic on
* `scored_candidates` - The candidates that the business logic layer is being asked to perform business logic on
## BusinessLogicResponse Objects
```python theme={null}
class BusinessLogicResponse(GenericModel, Generic[GENERALIZED_WYVERN_ENTITY,
REQUEST_ENTITY])
```
The response from the business logic layer after performing business logic on a set of candidates
**Arguments**:
* `request` - The request that the business logic layer was asked to perform business logic on
* `adjusted_candidates` - The candidates that the business logic layer performed business logic on
## BusinessLogicComponent Objects
```python theme={null}
class BusinessLogicComponent(Component[
BusinessLogicRequest[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY],
List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]],
], Generic[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY])
```
A component that performs business logic on an entity with a set of candidates
The request itself could contain more than just entities, for example it may contain a query and so on
## BusinessLogicPipeline Objects
```python theme={null}
class BusinessLogicPipeline(Component[
BusinessLogicRequest[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY],
BusinessLogicResponse[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY],
], Generic[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY])
```
Steps through a series of business logic components and returns the final output
This operation is fully chained, meaning that the output of each business logic component is passed
as an input to the next business logic component
#### execute
```python theme={null}
@tracer.wrap(name="BusinessLogicPipeline.execute")
async def execute(
input: BusinessLogicRequest[GENERALIZED_WYVERN_ENTITY,
REQUEST_ENTITY], **kwargs
) -> BusinessLogicResponse[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY]
```
Executes the business logic pipeline on the inputted candidates
**Arguments**:
* `input` - The input to the business logic pipeline
**Returns**:
The output of the business logic pipeline
#### extract\_business\_logic\_events
```python theme={null}
def extract_business_logic_events(
output: List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]],
pipeline_index: int, upstream_name: str, request_id: str,
old_scores: List[float]) -> List[BusinessLogicEvent]
```
Extracts the business logic events from the output of a business logic component
**Arguments**:
* `output` - The output of a business logic component
* `pipeline_index` - The index of the business logic component in the business logic pipeline
* `upstream_name` - The name of the business logic component
* `request_id` - The request id of the request that the business logic component was called in
* `old_scores` - The old scores of the candidates that the business logic component was called on
**Returns**:
The business logic events that were extracted from the output of the business logic component
# wyvern.components.business\_logic.pinning\_business\_logic
## PinningBusinessLogicComponent Objects
```python theme={null}
class PinningBusinessLogicComponent(
BusinessLogicComponent[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY],
Generic[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY])
```
A component that performs boosting on an entity with a set of candidates
The request itself could contain more than just entities, for example it may contain a query and so on
#### pin
```python theme={null}
def pin(
scored_candidates: List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]],
entity_pins: Dict[str, int],
entity_key_mapping: Callable[
[GENERALIZED_WYVERN_ENTITY],
str,
] = lambda candidate: candidate.identifier.identifier,
allow_down_ranking: bool = False
) -> List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]]
```
Pins the supplied entity to the specific position
**Arguments**:
* `scored_candidates` - The list of scored candidates
* `entity_pins` - The map of entity keys (unique identifiers) to pin, and their pinning position
* `entity_key_mapping` - A lambda function that takes in a candidate entity and
returns the field we should apply the pin to
* `allow_down_ranking` - Whether to allow down-ranking of candidates that are not pinned
**Returns**:
The list of scored candidates with the pinned entities
# wyvern.components.ranking\_pipeline
## RankingRequest Objects
```python theme={null}
class RankingRequest(BaseWyvernRequest, PaginationFields,
Generic[WYVERN_ENTITY])
```
This is the request for the ranking pipeline.
**Attributes**:
* `query` - the query entity
* `candidates` - the list of candidate entities
## ResponseCandidate Objects
```python theme={null}
class ResponseCandidate(BaseModel)
```
This is the response candidate.
**Attributes**:
* `candidate_id` - the identifier of the candidate
* `ranked_score` - the ranked score of the candidate
## RankingResponse Objects
```python theme={null}
class RankingResponse(BaseModel)
```
This is the response for the ranking pipeline.
**Attributes**:
* `ranked_candidates` - the list of ranked candidates
* `events` - the list of logged events
## RankingPipeline Objects
```python theme={null}
class RankingPipeline(PipelineComponent[RankingRequest, RankingResponse],
Generic[WYVERN_ENTITY])
```
This is the ranking pipeline.
**Attributes**:
* `PATH` - the path of the API. This is used in the API routing. The default value is "/ranking".
#### get\_model
```python theme={null}
def get_model() -> ModelComponent
```
This is the ranking model.
The model input should be a subclass of ModelInput.
Its output should be scored candidates
#### get\_business\_logic
```python theme={null}
def get_business_logic() -> Optional[BusinessLogicPipeline]
```
This is the business logic pipeline. It is optional. If not provided, the ranking pipeline will not
apply any business logic.
The business logic pipeline should be a subclass of BusinessLogicPipeline. Some examples of business logic
for ranking pipeline are:
1. Deduplication
2. Filtering
3. (De)boosting
#### rank\_candidates
```python theme={null}
async def rank_candidates(
request: RankingRequest[WYVERN_ENTITY]
) -> List[ScoredCandidate[WYVERN_ENTITY]]
```
This function ranks the candidates.
1. It first calls the ranking model to get the model scores for the candidates.
2. It then calls the business logic pipeline to adjust the model scores.
3. It returns the adjusted candidates.
**Arguments**:
* `request` - the ranking request
**Returns**:
A list of ScoredCandidate
# wyvern.components.index.\_index
## IndexUploadComponent Objects
```python theme={null}
class IndexUploadComponent(APIRouteComponent[IndexRequest, IndexResponse])
```
#### execute
```python theme={null}
async def execute(input: IndexRequest, **kwargs) -> IndexResponse
```
bulk index entities with redis pipeline
# wyvern.components.index
# wyvern.components.impressions
# wyvern.components.impressions.impression\_logger
## ImpressionEventData Objects
```python theme={null}
class ImpressionEventData(EntityEventData)
```
Impression event data. This is the data that is logged for each impression.
**Arguments**:
* `impression_score` - The score of the impression.
* `impression_order` - The order of the impression.
## ImpressionEvent Objects
```python theme={null}
class ImpressionEvent(LoggedEvent[ImpressionEventData])
```
Impression event. This is the event that is logged for each impression.
**Arguments**:
* `event_type` - The type of the event. This is always EventType.IMPRESSION.
## ImpressionEventLoggingRequest Objects
```python theme={null}
class ImpressionEventLoggingRequest(GenericModel,
Generic[GENERALIZED_WYVERN_ENTITY,
REQUEST_ENTITY])
```
Impression event logging request.
**Arguments**:
* `request` - The request that was made.
* `scored_impressions` - The scored impressions. This is a list of scored candidates.
Each scored candidate has an entity and a score.
## ImpressionEventLoggingComponent Objects
```python theme={null}
class ImpressionEventLoggingComponent(Component[
ImpressionEventLoggingRequest[GENERALIZED_WYVERN_ENTITY,
REQUEST_ENTITY],
None,
], Generic[GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY])
```
Impression event logging component. This component logs impression events.
#### execute
```python theme={null}
@tracer.wrap(name="ImpressionEventLoggingComponent.execute")
async def execute(input: ImpressionEventLoggingRequest[
GENERALIZED_WYVERN_ENTITY, REQUEST_ENTITY], **kwargs) -> None
```
Logs impression events.
**Arguments**:
* `input` - The input to the component. This contains the request and the scored impressions.
**Returns**:
None
# wyvern.components.events.events
## EventType Objects
```python theme={null}
class EventType(str, Enum)
```
Enum for the different types of events that can be logged.
## LoggedEvent Objects
```python theme={null}
class LoggedEvent(GenericModel, Generic[EVENT_DATA])
```
Base class for all logged events.
**Attributes**:
* `request_id` - The request ID of the request that triggered the event.
* `api_source` - The API source of the request that triggered the event.
* `event_timestamp` - The timestamp of the event.
* `event_type` - The type of the event.
* `event_data` - The data associated with the event. This is a generic type that can be any subclass of BaseModel.
## EntityEventData Objects
```python theme={null}
class EntityEventData(BaseModel)
```
Base class for all entity event data.
**Attributes**:
* `entity_identifier` - The identifier of the entity that the event is associated with.
* `entity_identifier_type` - The type of the entity identifier.
## CustomEvent Objects
```python theme={null}
class CustomEvent(LoggedEvent[ENTITY_EVENT_DATA_TYPE])
```
Class for custom events. Custom event data must be a subclass of EntityEventData.
**Attributes**:
* `event_type` - The type of the event. This is always EventType.CUSTOM.
# wyvern.components.events
# wyvern.components.component
## ComponentStatus Objects
```python theme={null}
class ComponentStatus(str, Enum)
```
This enum defines the status of the component.
## Component Objects
```python theme={null}
class Component(Generic[INPUT_TYPE, OUTPUT_TYPE])
```
Component is the base class for all the components in Wyvern. It is a generic class that takes in
the input type and the output type of the component.
It is responsible for: 1. Initializing the component 2. Initializing the upstream components
#### initialize
```python theme={null}
async def initialize() -> None
```
This is the place where you can do some initialization work for your component
As an example, you can initialize a model here or load a file,
which is needed for your component to work
#### initialize\_wrapper
```python theme={null}
async def initialize_wrapper() -> None
```
Extend this method if your component has some work that needs to be done on server startup
This is a great place to initialize libraries to access external libraries, warm up models, etc
This runs after all objects have been constructed
#### execute
```python theme={null}
async def execute(input: INPUT_TYPE, **kwargs) -> OUTPUT_TYPE
```
The actual meat of the component.
Custom component has to implement
If your component has to complex input data structure, make sure to override this method in order to
construct your input data with upstream components' output data
upstream\_outputs contains data that was parsed by upstreams
#### manifest\_feature\_names
```python theme={null}
@cached_property
def manifest_feature_names() -> Set[str]
```
This function defines which features are required for this component to work
Our system will automatically fetch the required features from the feature store
to make this model evaluation possible
#### get\_feature
```python theme={null}
def get_feature(identifier: Identifier, feature_name: str) -> WyvernFeature
```
This function gets the feature value for the given identifier
The features are cached once fetched/evaluated.
The feature that lives in the feature store should be
just using the feature name without the "feature\_view:" prefix
For example, if your you have a feature view "fv" and a feature "wyvern\_feature",
then you would have defined "fv:wyvern\_feature" in manifest\_feature\_names.
However, when you fetch the feature value with this function,
you just have to pass in feature\_name="wyvern\_feature".
#### get\_all\_features
```python theme={null}
def get_all_features(identifier: Identifier) -> Dict[str, WyvernFeature]
```
This function gets all features for the given identifier
The features are cached once fetched/evaluated.
# wyvern.components.pipeline\_component
## PipelineComponent Objects
```python theme={null}
class PipelineComponent(APIRouteComponent[REQUEST_ENTITY, RESPONSE_SCHEMA])
```
PipelineComponent is the base class for all the pipeline components in Wyvern. It is a Component that
takes in a request entity and a response schema, and it is responsible for hydrating the request
data with Wyvern Index data, and then pass the hydrated data to the next component in the pipeline.
#### realtime\_features\_overrides
```python theme={null}
@cached_property
def realtime_features_overrides() -> Set[Type[RealtimeFeatureComponent]]
```
This function defines the set of RealtimeFeatureComponents that generates features
with non-deterministic feature names.
For example, feature names like matched*query\_brand.
That feature is defined like matched\_query*{input.query.matched_query}, so it can refer to 10 or 20 features
#### retrieve\_features
```python theme={null}
async def retrieve_features(request: REQUEST_ENTITY) -> None
```
TODO shu: it doesn't support feature overrides. Write code to support that
# wyvern.components.helpers.linear\_algebra
## CosineSimilarityComponent Objects
```python theme={null}
class CosineSimilarityComponent(Component[List[Tuple[List[float],
List[float]]],
List[float]])
```
A component that computes cosine similarity in parallel for all pairs of embeddings.
#### execute
```python theme={null}
async def execute(input: List[Tuple[List[float], List[float]]],
**kwargs) -> List[float]
```
Computes cosine similarity in parallel for all pairs of embeddings.
**Arguments**:
* `input` - List of tuples of embeddings to compute cosine similarity for.
**Returns**:
List of cosine similarities.
#### cosine\_similarity
```python theme={null}
async def cosine_similarity(embedding_1: List[float],
embedding_2: List[float]) -> float
```
Computes cosine similarity between two embeddings.
# wyvern.components.helpers.sorting
## SortingComponent Objects
```python theme={null}
class SortingComponent(Component[
List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]],
List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]],
])
```
Sorts a list of candidates based on a score.
#### execute
```python theme={null}
async def execute(
input: List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]],
descending=True,
**kwargs) -> List[ScoredCandidate[GENERALIZED_WYVERN_ENTITY]]
```
Sorts a list of candidates based on a score.
**Arguments**:
* `input` - A list of candidates to be sorted. Each candidate must have a score.
* `descending` - Whether to sort in descending order. Defaults to True.
**Returns**:
A sorted list of candidates.
# wyvern.components.helpers
# wyvern.redis
## WyvernRedis Objects
```python theme={null}
class WyvernRedis()
```
WyvernRedis is a wrapper for redis client to help index your entities in redis with Wyvern's convention
#### \_\_init\_\_
```python theme={null}
def __init__(scope: str = "",
redis_host: Optional[str] = None,
redis_port: Optional[int] = None) -> None
```
scope is used to prefix the redis key. You can use the environment variable PROJECT\_NAME to set the scope.
#### get\_entity
```python theme={null}
async def get_entity(entity_type: str,
entity_id: str) -> Optional[Dict[str, Any]]
```
get entity from redis
#### get\_entities
```python theme={null}
async def get_entities(
entity_type: str,
entity_ids: Sequence[str]) -> List[Optional[Dict[str, Any]]]
```
get entity from redis
#### delete\_entity
```python theme={null}
async def delete_entity(entity_type: str, entity_id: str) -> None
```
delete entity from redis
#### delete\_entities
```python theme={null}
async def delete_entities(entity_type: str, entity_ids: Sequence[str]) -> None
```
delete entities from redis
# wyvern.utils
# wyvern.wyvern\_tracing
#### setup\_tracing
```python theme={null}
def setup_tracing()
```
Setup tracing for Wyvern service. Tracing is disabled in development mode and for healthcheck requests.
# wyvern.wyvern\_logging
#### setup\_logging
```python theme={null}
def setup_logging()
```
Setup logging configuration by loading from log\_config.yml file. Logs an error if the
file cannot be found or loaded and uses default logging configuration.
# wyvern.exceptions
## WyvernError Objects
```python theme={null}
class WyvernError(Exception)
```
Base class for all Wyvern errors.
**Attributes**:
* `message` - The error message.
* `error_code` - The error code.
## WyvernEntityValidationError Objects
```python theme={null}
class WyvernEntityValidationError(WyvernError)
```
Raised when entity data is invalid
## PaginationError Objects
```python theme={null}
class PaginationError(WyvernError)
```
Raised when there is an error in pagination
## WyvernRouteRegistrationError Objects
```python theme={null}
class WyvernRouteRegistrationError(WyvernError)
```
Raised when there is an error in registering a route
## ComponentAlreadyDefinedInPipelineComponentError Objects
```python theme={null}
class ComponentAlreadyDefinedInPipelineComponentError(WyvernError)
```
Raised when a component is already defined in a pipeline component
## WyvernFeatureStoreError Objects
```python theme={null}
class WyvernFeatureStoreError(WyvernError)
```
Raised when there is an error in feature store
## WyvernFeatureNameError Objects
```python theme={null}
class WyvernFeatureNameError(WyvernError)
```
Raised when there is an error in feature name
## WyvernModelInputError Objects
```python theme={null}
class WyvernModelInputError(WyvernError)
```
Raised when there is an error in model input
## WyvernModelbitTokenMissingError Objects
```python theme={null}
class WyvernModelbitTokenMissingError(WyvernError)
```
Raised when modelbit token is missing
## WyvernModelbitValidationError Objects
```python theme={null}
class WyvernModelbitValidationError(WyvernError)
```
Raised when modelbit validation fails
## WyvernAPIKeyMissingError Objects
```python theme={null}
class WyvernAPIKeyMissingError(WyvernError)
```
Raised when api key is missing
## ExperimentationProviderNotSupportedError Objects
```python theme={null}
class ExperimentationProviderNotSupportedError(WyvernError)
```
Raised when experimentation provider is not supported
## ExperimentationClientInitializationError Objects
```python theme={null}
class ExperimentationClientInitializationError(WyvernError)
```
Raised when experimentation client initialization fails
# wyvern.aws.kinesis
## KinesisFirehoseStream Objects
```python theme={null}
class KinesisFirehoseStream(str, Enum)
```
Enum for Kinesis Firehose stream names
Usage:
```
>>> KinesisFirehoseStream.EVENT_STREAM.get_stream_name()
```
#### get\_stream\_name
```python theme={null}
def get_stream_name(customer_specific: bool = True,
env_specific: bool = True) -> str
```
Returns the stream name for the given stream
**Arguments**:
* `customer_specific` - Whether the stream name should be customer specific
* `env_specific` - Whether the stream name should be environment specific
**Returns**:
The stream name
## WyvernKinesisFirehose Objects
```python theme={null}
class WyvernKinesisFirehose()
```
Wrapper around boto3 Kinesis Firehose client
#### put\_record\_batch\_callable
```python theme={null}
def put_record_batch_callable(
stream_name: KinesisFirehoseStream,
record_generator: List[Callable[[], List[BaseModel]]])
```
Puts records to the given stream. This is a callable that can be used with FastAPI's BackgroundTasks. This
way events can be logged asynchronously after the response is sent to the client.
**Arguments**:
* `stream_name` *KinesisFirehoseStream* - The stream to put records to
* `record_generator` *List\[Callable\[\[], List\[BaseModel]]]* - A list of functions that return a list of records
**Returns**:
None
#### put\_record\_batch
```python theme={null}
def put_record_batch(stream_name: KinesisFirehoseStream,
records: List[BaseModel])
```
Puts records to the given stream
**Arguments**:
* `stream_name` *KinesisFirehoseStream* - The stream to put records to
* `records` *List\[BaseModel]* - A list of records
**Returns**:
None
# wyvern.aws
# wyvern.event\_logging
# wyvern.event\_logging.event\_logger
#### log\_events
```python theme={null}
def log_events(event_generator: Callable[[], List[LoggedEvent]])
```
Logs events to the current request context.
**Arguments**:
* `event_generator` - A function that returns a list of events to be logged.
#### get\_logged\_events
```python theme={null}
def get_logged_events() -> List[LoggedEvent[Any]]
```
**Returns**:
A list of all the events logged in the current request context.
#### get\_logged\_events\_generator
```python theme={null}
def get_logged_events_generator(
) -> List[Callable[[], List[LoggedEvent[Any]]]]
```
**Returns**:
A list of all the event generators logged in the current request context.
#### log\_custom\_events
```python theme={null}
def log_custom_events(events: List[ENTITY_EVENT_DATA_TYPE]) -> None
```
Logs custom events to the current request context.
**Arguments**:
* `events` - A list of custom events to be logged.
# wyvern.web\_frameworks.fastapi
#### lifespan
```python theme={null}
@asynccontextmanager
async def lifespan(app: FastAPI)
```
A context manager that starts and stops with the app. This is used to start and stop the aiohttp client.
## WyvernFastapi Objects
```python theme={null}
class WyvernFastapi()
```
A wrapper around FastAPI that provides a few additional features:
* A healthcheck endpoint
* A request middleware that logs the request and response payloads
* A request middleware that sets the WyvernRequest in the request context
* Auto registration of routes from APIRouteComponent subclasses
endpoint input:
the built WyvernPipeline
the request input schema
the request output schema
#### register\_route
```python theme={null}
async def register_route(route_component: Type[APIRouteComponent]) -> None
```
Register a route component. This will register the route with FastAPI and also initialize the route component.
**Arguments**:
* `route_component` - The route component to register.
**Raises**:
* `WyvernRouteRegistrationError` - If the route component is not a subclass of APIRouteComponent.
# wyvern.web\_frameworks
# wyvern.wyvern\_request
## WyvernRequest Objects
```python theme={null}
@dataclass
class WyvernRequest()
```
WyvernRequest is a dataclass that represents a request to the Wyvern service. It is used to pass
information between the various components of the Wyvern service.
**Attributes**:
* `method` - The HTTP method of the request
* `url` - The full URL of the request
* `url_path` - The path of the URL of the request
* `json` - The JSON body of the request, represented by pydantic model
* `headers` - The headers of the request
* `entity_store` - A dictionary that can be used to store entities that are created during the request
* `events` - A list of functions that return a list of LoggedEvents. These functions are called at the end of
the request to log events to the event store
* `feature_map` - A FeatureMap that can be used to store features that are created during the request
* `request_id` - The request ID of the request
#### parse\_fastapi\_request
```python theme={null}
@classmethod
def parse_fastapi_request(cls,
json: BaseModel,
req: fastapi.Request,
request_id: Optional[str] = None) -> WyvernRequest
```
Parses a FastAPI request into a WyvernRequest
**Arguments**:
* `json` - The JSON body of the request, represented by pydantic model
* `req` - The FastAPI request
* `request_id` - The request ID of the request
**Returns**:
A WyvernRequest
# wyvern.tracking
# wyvern.feature\_store.constants
# wyvern.feature\_store
# wyvern.feature\_store.schemas
## GetOnlineFeaturesRequest Objects
```python theme={null}
class GetOnlineFeaturesRequest(BaseModel)
```
Request object for getting online features.
**Attributes**:
* `entities` - A dictionary of entity name to entity value.
* `features` - A list of feature names.
* `full_feature_names` - A boolean indicating whether to return full feature names. If True, the feature names will
be returned in the format `__`. If False, only the feature names will be
returned.
## GetHistoricalFeaturesRequest Objects
```python theme={null}
class GetHistoricalFeaturesRequest(BaseModel)
```
Request object for getting historical features.
**Attributes**:
* `entities` - A dictionary of entity name to entity value.
* `timestamps` - A list of timestamps. Used to retrieve historical features at specific timestamps. If not provided,
the latest feature values will be returned.
* `features` - A list of feature names.
## GetFeastHistoricalFeaturesRequest Objects
```python theme={null}
class GetFeastHistoricalFeaturesRequest(BaseModel)
```
Request object for getting historical features from Feast.
**Attributes**:
* `full_feature_names` - A boolean indicating whether to return full feature names. If True, the feature names will
be returned in the format `__`. If False, only the feature names will be
returned.
* `entities` - A dictionary of entity name to entity value.
* `features` - A list of feature names.
## GetHistoricalFeaturesResponse Objects
```python theme={null}
class GetHistoricalFeaturesResponse(BaseModel)
```
Response object for getting historical features.
**Attributes**:
* `results` - A list of dictionaries containing feature values.
## MaterializeRequest Objects
```python theme={null}
class MaterializeRequest(BaseModel)
```
Request object for materializing feature views.
**Attributes**:
* `end_date` - The end date of the materialization window. Defaults to the current time.
* `feature_views` - A list of feature view names to materialize. If not provided, all feature views will be
materialized.
* `start_date` - The start date of the materialization window. Defaults to None, which will use the start date of
the feature view.
## RequestEntityIdentifierObjects Objects
```python theme={null}
class RequestEntityIdentifierObjects(BaseModel)
```
Request object for getting entity identifier objects.
**Attributes**:
* `request_ids` - A list of request IDs.
* `entity_identifiers` - A list of entity identifiers.
* `feature_names` - A list of feature names.
# wyvern.feature\_store.feature\_server
#### CRONJOB\_INTERVAL\_SECONDS
5 minutes
#### CRONJOB\_LOOKBACK\_MINUTES
12 mins
#### generate\_wyvern\_store\_app
```python theme={null}
def generate_wyvern_store_app(path: str) -> FastAPI
```
Generate a FastAPI app for Wyvern feature store.
**Arguments**:
* `path` - Path to the feature store repo.
**Returns**:
FastAPI app.
#### start\_wyvern\_store
```python theme={null}
def start_wyvern_store(path: str, host: str, port: int)
```
Start the Wyvern feature store.
**Arguments**:
* `path` - Path to the feature store repo.
* `host` - Host to run the feature store on.
* `port` - Port to run the feature store on.
# wyvern.feature\_store.historical\_feature\_util
#### separate\_real\_time\_features
```python theme={null}
def separate_real_time_features(
full_feature_names: Optional[List[str]]
) -> Tuple[List[str], List[str]]
```
Given a list of full feature names, separate real-time features and other features.
**Arguments**:
* `full_feature_names` - a list of full feature names.
**Returns**:
Real time feature names and other feature names in two lists respectively.
#### build\_historical\_real\_time\_feature\_requests
```python theme={null}
def build_historical_real_time_feature_requests(
full_feature_names: List[str], request_ids: List[str],
entities: Dict[str,
List[Any]]) -> Dict[str, RequestEntityIdentifierObjects]
```
Build historical real-time feature requests grouped by entity types so that we can process them in parallel.
**Arguments**:
* `full_feature_names` - a list of full feature names.
* `request_ids` - a list of request ids.
* `entities` - a dictionary of entity names and their values.
**Returns**:
A dictionary of entity types and their corresponding requests.
#### process\_historical\_real\_time\_features\_requests
```python theme={null}
def process_historical_real_time_features_requests(
requests: Dict[str, RequestEntityIdentifierObjects]
) -> Dict[str, pd.DataFrame]
```
Given a dictionary of historical real-time feature requests, process them and return the results.
**Arguments**:
* `requests` - a dictionary of entity types and their corresponding requests.
**Returns**:
A dictionary of entity types and their corresponding results in pandas dataframes.
#### process\_historical\_real\_time\_features\_request
```python theme={null}
def process_historical_real_time_features_request(
entity_identifier_type: str, request: RequestEntityIdentifierObjects,
context: SnowflakeConnection) -> pd.DataFrame
```
Given a historical real-time feature request, process it and return the results.
**Arguments**:
* `entity_identifier_type` - the entity type of the request. E.g. "product\_\_query"
* `request` - the request object.
* `context` - the snowflake connection context.
**Returns**:
The result in pandas dataframe.
#### group\_realtime\_features\_by\_entity\_type
```python theme={null}
def group_realtime_features_by_entity_type(
full_feature_names: List[str]) -> Dict[str, List[str]]
```
Given a list of feature names, group them by their entity\_identifier\_type
**Arguments**:
* `full_feature_names` - a list of full feature names.
**Returns**:
A dictionary of entity types and their corresponding feature names.
#### group\_registry\_features\_by\_entities
```python theme={null}
def group_registry_features_by_entities(
full_feature_names: List[str],
store: FeatureStore) -> Dict[str, List[str]]
```
Given a list of feature names, group them by their entity name.
**Arguments**:
* `full_feature_names` - a list of full feature names.
* `store` - the feast feature store.
**Returns**:
A dictionary of entity names and their corresponding feature names.
#### build\_historical\_registry\_feature\_requests
```python theme={null}
def build_historical_registry_feature_requests(
store: FeatureStore, feature_names: List[str],
entity_values: Dict[str, List[Any]],
timestamps: List[datetime]) -> List[GetFeastHistoricalFeaturesRequest]
```
Build historical feature requests grouped by entity names so that we can process them in parallel.
**Arguments**:
* `store` - the feast feature store.
* `feature_names` - a list of feature names.
* `entity_values` - a dictionary of entity names and their values.
* `timestamps` - a list of timestamps for getting historical features at those timestamps.
**Returns**:
A list of historical feature requests.
#### process\_historical\_registry\_features\_requests
```python theme={null}
def process_historical_registry_features_requests(
store: FeatureStore, requests: List[GetFeastHistoricalFeaturesRequest]
) -> List[pd.DataFrame]
```
Given a list of historical feature requests, process them and return the results
**Arguments**:
* `store` - the feast feature store.
* `requests` - a list of historical feature requests.
**Returns**:
A list of results in pandas dataframes.
#### process\_historical\_registry\_features\_request
```python theme={null}
def process_historical_registry_features_request(
store: FeatureStore,
request: GetFeastHistoricalFeaturesRequest) -> pd.DataFrame
```
Given a historical feature request, process it and return the results
**Arguments**:
* `store` - the feast feature store.
* `request` - a historical feature request.
**Returns**:
The result in pandas dataframe.
# wyvern.helper.sort
## SortEnum Objects
```python theme={null}
class SortEnum(str, Enum)
```
Enum for sort order.
## Sort Objects
```python theme={null}
class Sort(BaseModel)
```
Sort class for sorting the results.
**Attributes**:
* `sort_key` - The key to sort on.
* `sort_field` - The field to sort on.
* `sort_order` - The order to sort on. Defaults to desc.
# wyvern.helper
# wyvern.entities.feature\_entity\_helpers
#### feature\_map\_join
```python theme={null}
def feature_map_join(*feature_maps: FeatureMap) -> FeatureMap
```
Joins multiple feature maps into a single feature map. Used to join feature maps from different sources.
#### feature\_map\_create
```python theme={null}
def feature_map_create(*feature_data: Optional[FeatureData]) -> FeatureMap
```
Creates a feature map from a list of feature data. Used to create feature maps from different sources.
# wyvern.entities.index\_entities
# wyvern.entities.identifier\_entities
## WyvernDataModel Objects
```python theme={null}
class WyvernDataModel(BaseModel)
```
WyvernDataModel is a base class for all data models that could be hydrated from Wyvern Index.
**Attributes**:
* `_all_entities` - a list of all the entities under the tree
* `_all_identifiers` - a list of all the identifiers under the tree
#### index\_fields
```python theme={null}
def index_fields() -> List[str]
```
This method returns a list of fields that contains indexable data
#### get\_all\_entities
```python theme={null}
def get_all_entities(cached: bool = True) -> List[WyvernEntity]
```
This method returns all of the entities associated with subclasses of this
If cached is True, all the nodes under the tree will be cached
#### get\_all\_identifiers
```python theme={null}
def get_all_identifiers(cached: bool = True) -> List[Identifier]
```
This method generally returns all of the identifiers associated with subclasses of this
Example: You create a QueryProductEntity with query="test" and product\_id="1234"
It subclasses QueryEntity and ProductEntity, which both have an identifier
This method will return a list of both of those identifiers
Example: You create a ProductSearchRankingRequest with
query="test", candidates=\["1234", ...], user="u\_1234"
This method will return the user and query identifier
It will also return the identifiers for each candidate (thanks to the implementation in CandidateEntity)
Note: While this checks for `WyvernEntity` -- a `WyvernDataModel` can have many
entities within it, it itself may not be an entity
#### nested\_hydration
```python theme={null}
def nested_hydration() -> Dict[str, str]
```
A dictionary that maps the entity id field name to the nested entity field name
TODO: \[SHU] replace this mapping by introducing `class WyvernField(pydantic.Field)`
to represent the "entity ide field", which will reference to the nested entity field name
## WyvernEntity Objects
```python theme={null}
class WyvernEntity(WyvernDataModel)
```
WyvernEntity is a base class for all entities that have primary identifier.
An entity is the basic unit of data that could be indexed and queried.
#### identifier
```python theme={null}
@property
def identifier() -> Identifier
```
This method returns the identifier for this entity
#### load\_fields
```python theme={null}
def load_fields(data: Dict[str, Any]) -> None
```
This method load the entity with the given data.
The return data is the nested entities that need to be further hydrated
For example:
if a Product contains these two fields: `brand_id: Optional[str]` and `brand: Optional[Brand]`,
as the hydrated entity. We fetch the brand\_id for the product from Wyvern Index,
as the first hydration step for Product entity, then we fetch brand entity from Wyvern Index,
as the second hydration step
## QueryEntity Objects
```python theme={null}
class QueryEntity(WyvernEntity)
```
QueryEntity is a base class for all entities that have query as an identifier.
**Attributes**:
* `query` - the query string
#### generate\_identifier
```python theme={null}
def generate_identifier() -> Identifier
```
This method returns the identifier for this entity.
**Returns**:
* `Identifier` - the identifier for this entity with identifier\_type=SimpleIdentifierType.QUERY.
## ProductEntity Objects
```python theme={null}
class ProductEntity(WyvernEntity)
```
ProductEntity is a base class for all entities that have product\_id as an identifier.
**Attributes**:
* `product_id` - the product id
#### generate\_identifier
```python theme={null}
def generate_identifier() -> Identifier
```
This method returns the identifier for this entity.
**Returns**:
* `Identifier` - the identifier for this entity with identifier\_type=SimpleIdentifierType.PRODUCT.
## UserEntity Objects
```python theme={null}
class UserEntity(WyvernEntity)
```
UserEntity is a base class for all entities that have user\_id as an identifier.
**Attributes**:
* `user_id` - the user id
#### generate\_identifier
```python theme={null}
def generate_identifier() -> Identifier
```
This method returns the identifier for this entity.
**Returns**:
* `Identifier` - the identifier for this entity with identifier\_type=SimpleIdentifierType.USER.
# wyvern.entities.request
## BaseWyvernRequest Objects
```python theme={null}
class BaseWyvernRequest(WyvernDataModel)
```
Base class for all Wyvern requests. This class is used to generate an identifier for the request.
**Attributes**:
* `request_id` - The request id.
* `include_events` - Whether to include events in the response.
#### generate\_identifier
```python theme={null}
def generate_identifier() -> Identifier
```
Generates an identifier for the request.
**Returns**:
* `Identifier` - The identifier for the request. The identifier type is "request".
# wyvern.entities
# wyvern.entities.feature\_entities
## FeatureData Objects
```python theme={null}
class FeatureData(BaseModel)
```
A class to represent the features of an entity.
**Attributes**:
* `identifier` - The identifier of the entity.
* `features` - A dictionary of feature names to feature values.
## FeatureMap Objects
```python theme={null}
class FeatureMap(BaseModel)
```
A class to represent a map of identifiers to feature data.
TODO (kerem): Fix the data duplication between this class and the FeatureData class. The identifier field in the
FeatureData class is redundant.
#### build\_empty\_feature\_map
```python theme={null}
def build_empty_feature_map(identifiers: List[Identifier],
feature_names: List[str]) -> FeatureMap
```
Builds an empty feature map with the given identifiers and feature names.
# wyvern.entities.identifier
## SimpleIdentifierType Objects
```python theme={null}
class SimpleIdentifierType(str, Enum)
```
Simple identifier types are those that are not composite.
#### composite
```python theme={null}
def composite(primary_identifier_type: SimpleIdentifierType,
secondary_identifier_type: SimpleIdentifierType) -> str
```
Composite identifier types are those that are composite. For example, a product with id p\_1234 and type "product"
a user with id u\_1234 and type "user" would have a composite identifier of "p\_1234:u\_1234", and a composite
identifier\_type of "product:user". This is useful for indexing and searching for composite entities.
## CompositeIdentifierType Objects
```python theme={null}
class CompositeIdentifierType(str, Enum)
```
Composite identifier types are those that are composite. For example, a composite identifier type of
"product:user" would be a composite identifier type for a product and a user. This is useful for indexing and
searching for composite entities.
## Identifier Objects
```python theme={null}
class Identifier(BaseModel)
```
Identifiers exist to represent a unique entity through their unique id and their type
For example: a product with id p\_1234 and type "product" or a user with id u\_1234 and type "user"
Composite identifiers are also possible, for example:
a product with id p\_1234 and type "product"
a user with id u\_1234 and type "user"
The composite identifier would be "p\_1234:u\_1234",
and the composite identifier\_type would be "product:user"
## CompositeIdentifier Objects
```python theme={null}
class CompositeIdentifier(Identifier)
```
Composite identifiers exist to represent a unique entity through their unique id and their type. At most, they
can have two identifiers and two identifier types. For example:
a product with id p\_1234 and type "product"
a user with id u\_1234 and type "user"
The composite identifier would be "p\_1234:u\_1234", and the composite identifier\_type would be "product:user".
# wyvern.entities.candidate\_entities
## ScoredCandidate Objects
```python theme={null}
class ScoredCandidate(GenericModel, Generic[GENERALIZED_WYVERN_ENTITY])
```
A candidate entity with a score.
**Attributes**:
* `entity` - The candidate entity.
* `score` - The score of the candidate entity. Defaults to 0.0.
## CandidateSetEntity Objects
```python theme={null}
class CandidateSetEntity(WyvernDataModel, GenericModel,
Generic[GENERALIZED_WYVERN_ENTITY])
```
A set of candidate entities. This is a generic model that can be used to represent a set of candidate entities.
**Attributes**:
* `candidates` - The list of candidate entities.
# examples
# examples.feature\_store\_main
#### run
```python theme={null}
@wyvern_cli_app.command()
def run(host: str = "127.0.0.1", port: int = 8000) -> None
```
Run your wyvern service
# examples.example\_business\_logic
#### sample\_product\_query\_ranking\_request
```python theme={null}
async def sample_product_query_ranking_request() -> None
```
How to run this: `python wyvern/examples/example_business_logic.py`
Json representation of the request:
```
{
"request_id": "rrr",
"query": "candle",
"candidates": [
{"product_id": "1", "product_name": "scented candle"},
{"product_id": "2", "product_name": "hot candle"},
{"product_id": "3", "product_name": "pumpkin candle"},
{"product_id": "4", "product_name": "unrelated item"},
{"product_id": "5", "product_name": "candle holder accessory"},
{"product_id": "6", "product_name": "earwax holder"},
{"product_id": "7", "product_name": "wax seal"}
],
}
```
# examples.real\_time\_features\_main
#### run
```python theme={null}
@wyvern_cli_app.command()
def run(host: str = "127.0.0.1", port: int = 8000) -> None
```
Run your wyvern service