# trulens.providers.cortex.endpoint

## `trulens.providers.cortex.endpoint`

### Classes

#### `CortexCallback`

Bases: `EndpointCallback`

##### Attributes

###### `endpoint`
```python
endpoint: Endpoint = Field(exclude=True)
```
The endpoint owning this callback.

###### `cost`
```python
cost: Cost = Field(default_factory=Cost)
```
Costs tracked by this callback.

##### Functions

###### `handle_generation`
```python
handle_generation(response: dict) -> None
```
Get the usage information from Cortex LLM function response's usage field.

###### `__repr__`
```python
__repr__() -> str
```
Safe `repr` that handles circular references.

###### `__rich_repr__`
```python
__rich_repr__() -> Result
```
Requirement for pretty printing using the rich package.

###### `handle`
```python
handle(response: Any) -> None
```
Called after each request.

###### `handle_chunk`
```python
handle_chunk(response: Any) -> None
```
Called after receiving a chunk from a request.

###### `handle_generation_chunk`
```python
handle_generation_chunk(response: Any) -> None
```
Called after receiving a chunk from a completion request.

###### `handle_classification`
```python
handle_classification(response: Any) -> None
```
Called after each classification response.

###### `handle_embedding`
```python
handle_embedding(response: Any) -> None
```
Called after each embedding response.

#### `CortexEndpoint`

Bases: `Endpoint`

Snowflake Cortex endpoint.

##### Attributes

###### `tru_class_info`
```python
tru_class_info: Class
```
Class information of this pydantic object for use in deserialization.

###### `instrumented_methods`
```python
instrumented_methods: Dict[
    Any, List[Tuple[Callable, Callable, Type[Endpoint]]]
] = defaultdict(list)
```
Mapping of classes/module-methods that have been instrumented for cost tracking.

###### `name`
```python
name: str
```
API/endpoint name.

###### `rpm`
```python
rpm: float = DEFAULT_RPM
```
Requests per minute.

###### `retries`
```python
retries: int = 3
```
Retries (if performing requests using this class).

###### `post_headers`
```python
post_headers: Dict[str, str] = Field(
    default_factory=dict, exclude=True
)
```
Optional post headers for post requests if done by this class.

###### `pace`
```python
pace: Pace = Field(
    default_factory=lambda: Pace(
        marks_per_second=DEFAULT_RPM / 60.0,
        seconds_per_period=60.0,
    ),
    exclude=True,
)
```
Pacing instance to maintain a desired rpm.

###### `global_callback`
```python
global_callback: EndpointCallback = Field(exclude=True)
```
Track costs not run inside "track_cost" here.

###### `callback_class`
```python
callback_class: Type[EndpointCallback] = Field(exclude=True)
```
Callback class to use for usage tracking.

###### `callback_name`
```python
callback_name: str = Field(exclude=True)
```
Name of variable that stores the callback noted above.

##### Classes

###### `EndpointSetup`  
Class for storing supported endpoint information.

##### Functions

###### `get_instances`  
```python
get_instances() -> Generator[InstanceRefMixin]
```
Get all instances of the class.

###### `delete_instances`
```python
delete_instances()
```
Delete all instances of the class.

###### `__rich_repr__`
```python
__rich_repr__() -> Result
```
Requirement for pretty printing using the rich package.

###### `load`
```python
load(obj, *args, **kwargs)
```
Deserialize/load this object using the class information in tru_class_info.

###### `model_validate`
```python
model_validate(*args, **kwargs) -> Any
```
Deserialized a jsonized version of the app into the instance of the class it was serialized from.

###### `pace_me`
```python
pace_me() -> float
```
Block until we can make a request to this endpoint to keep pace with maximum rpm. Returns time in seconds since last call to this method returned.

###### `run_in_pace`
```python
run_in_pace(
    func: Callable[[A], B], *args, **kwargs
) -> B
```
Run the given `func` with the provided `args` and `kwargs` at pace with the endpoint-specified rpm. Failures will be retried `self.retries` times.

###### `run_me`
```python
run_me(thunk: Thunk[T]) -> T
```
DEPRECATED: Run the given thunk, returning its output, on pace with the api.

###### `print_instrumented`
```python
print_instrumented()
```
Print out all of the methods that have been instrumented for cost tracking.

###### `track_all_costs`
```python
track_all_costs(
    __func: CallableMaybeAwaitable[A, T],
    *args,
    with_openai: bool = True,
    with_hugs: bool = True,
    with_litellm: bool = True,
    with_bedrock: bool = True,
    with_cortex: bool = True,
    with_dummy: bool = True,
    **kwargs
) -> Tuple[T, Sequence[EndpointCallback]]
```
Track costs of all of the apis we can currently track, over the execution of thunk.

###### `track_all_costs_tally`
```python
track_all_costs_tally(
    __func: CallableMaybeAwaitable[A, T],
    *args,
    with_openai: bool = True,
    with_hugs: bool = True,
    with_litellm: bool = True,
    with_bedrock: bool = True,
    with_cortex: bool = True,
    with_dummy: bool = True,
    **kwargs
) -> Tuple[T, Thunk[Cost]]
```
Track costs of all of the apis we can currently track, over the execution of thunk.

###### `track_cost`
```python
track_cost(
    __func: CallableMaybeAwaitable[..., T], *args, **kwargs
) -> Tuple[T, EndpointCallback]
```
Tally only the usage performed within the execution of the given thunk.

###### `wrap_function`
```python
wrap_function(func)
```
Create a wrapper of the given function to perform cost tracking.

### Functions

#### `register_otel_cost_tracking`
```python
register_otel_cost_tracking() -> None
```
Register OTel-tracing cost instrumentation for the Cortex provider.