Core API Reference

This section documents the da_vinci (core runtime) package API.

ORM Module

Table Objects

class da_vinci.core.orm.table_object.TableObjectAttributeType(*values)

Bases: StrEnum

STRING = 'string'

NUMBER = 'number'

BOOLEAN = 'boolean'

DATETIME = 'datetime'

JSON = 'json'

JSON_STRING = 'json_string'

STRING_LIST = 'string_list'

NUMBER_LIST = 'number_list'

JSON_LIST = 'json_list'

JSON_STRING_LIST = 'json_string_list'

COMPOSITE_STRING = 'composite_string'

STRING_SET = 'string_set'

NUMBER_SET = 'number_set'

classmethod is_list(attribute_type)

Check if the attribute type is a list

Keyword Arguments:

check (attribute_type -- Attribute type to)

Return type:

bool

Returns:

bool

to_str()

Convert the attribute type to a string

Return type:

str

Returns:

str

class da_vinci.core.orm.table_object.TableObjectAttribute(name, attribute_type, argument_names=None, custom_exporter=None, custom_importer=None, description=None, dynamodb_key_name=None, default=None, exclude_from_dict=False, exclude_from_schema_description=False, is_indexed=True, optional=False)

Bases: object

__init__(name, attribute_type, argument_names=None, custom_exporter=None, custom_importer=None, description=None, dynamodb_key_name=None, default=None, exclude_from_dict=False, exclude_from_schema_description=False, is_indexed=True, optional=False)

Object representing an attribute of a TableObject

Keyword Arguments:

• attribute (attribute_type -- Type of the)

• attribute

• composite (argument_names -- The names of Python arguments that are merged into a) – string. This is required when the attribute type is COMPOSITE_STRING.

• DynamoDB (custom_importer -- Custom importer function, called whenever data is loaded from)

• DynamoDB

• primarily (description -- Description of the attribute, annotation) – used for LLM context.

• based (dynamodb_key_name -- Name of the DynamoDB key, defaults to the ORM naming convention) – on the attribute name

• set. (default -- Default value for the attribute, attribute is considered optional when this is) – Accepts a value or a callable that returns a value.

• to_dict() (exclude_from_dict -- Attribute is not added to a resulting Dict when calling)

• description (exclude_from_schema_description -- Attribute is not included in the table object schema)

• True (is_indexed -- Whether the attribute is able to be used to query with, defaults to)

• provided (optional -- Whether the attribute optional, defaults to False unless a default is)

static composite_string_value(values)

Return a full composite string value given a list of attribute values

Keyword Arguments:

value (values -- List of values to join into the full composite string)

Return type:

str

static default_dynamodb_key_name(name)

Convert a name to a DynamoDB key name

Keyword Arguments:

convert (name -- Name to)

Return type:

str

Returns:

str

static timestamp_to_datetime(timestamp)

Convert a timestamp string to a datetime

Keyword Arguments:

string (timestamp -- Timestamp)

Return type:

datetime

Returns:

datetime

static datetime_to_timestamp(dt)

Convert a datetime to a timestamp

Keyword Arguments:

Datetime (dt --)

Return type:

float

Returns:

float

schema_to_str()

Describe the schema for the attribute

Return type:

str

Returns:

str

property default: Any

property dynamodb_type_label: str

Get the DynamoDB type label for the attribute type

Returns:

str

dynamodb_value(value)

Convert a value to a DynamoDB supported value

Keyword Arguments:

convert (value -- Value to)

Return type:

Any

Returns:

Any

as_dynamodb_attribute(value)

Return the attribute as a DynamoDB attribute

Keyword Arguments:

convert (value -- Value to)

Return type:

dict | None

true_value(value)

Return the attribute value as a Python value

Return type:

Any

set_attribute(obj, value)

Set the attribute on an object

Keyword Arguments:

• on (obj -- Object to set the attribute)

• set (value -- Value to)

Return type:

None

from_dynamodb_attribute(value)

Return the attribute value as a Python value, run a custom importer if one was set

Return type:

Any

class da_vinci.core.orm.table_object.TableObject(**kwargs)

Bases: object

Base class for Table object definitions

Class Attributes:

attribute_lookup_prefix: Attribute lookup prefix, prefixes the attribute name when retrieving attributes attributes: List of attributes description: Description of the table object_name: Name of the object, defaults to the class name. This should be set when

dynamically defining table objects.

partition_key_attribute: Partition key attribute sort_key_attribute: Sort key attribute table_name: Name of the table ttl_attribute: Optional TTL attribute

Example

``` from uuid import uuid4

from da_vinci.core.orm.table_object import (

TableObject, TableObjectAttribute, TableObjectAttributeType,

)

class MyTableObject(TableObject):

partition_key_attribute = TableObjectAttribute(

name=’my_pk’, attribute_type=TableObjectAttributeType.STRING,

)

sort_key_attribute = TableObjectAttribute(

name=’my_sk’, attribute_type=TableObjectAttributeType.STRING, default=lambda: str(uuid4()),

)

table_name = ‘my_table’

attributes = [

TableObjectAttribute(

name=’created_on’, attribute_type=TableObjectAttributeType.DATETIME, default=lambda: datetime.now(),

),

]

```

partition_key_attribute: TableObjectAttribute

table_name: str

attribute_lookup_prefix: str | None = None

attributes: list[TableObjectAttribute] = []

description: str | None = None

object_name: str | None = None

sort_key_attribute: TableObjectAttribute | None = None

ttl_attribute: TableObjectAttribute | None = None

__init__(**kwargs)

Base class for Table objects

classmethod define(partition_key_attribute, object_name, table_name, attribute_lookup_prefix=None, attributes=None, description=None, sort_key_attribute=None, ttl_attribute=None)

Define a TableObject

Keyword Arguments:

• attribute_lookup_prefix – Attribute lookup prefix

• attributes – List of attributes

• description – Description of the table

• partition_key_attribute – Partition key attribute

• object_name – Name of the object

• sort_key_attribute – Sort key attribute

• table_name – Name of the table

• ttl_attribute – Optional TTL attribute

Return type:

type[TableObject]

__getattr__(name)

Get an attribute by name

Keyword Arguments:

attribute (name -- Name of the)

Return type:

Any

Returns:

Any

attribute_value(name)

Get the value of an attribute

Keyword arguments: name – Name of the attribute

Return type:

Any

composite_attribute_values(name)

Get a dictionary representation of the composite attribute’s values

Keyword arguments: name – Name of the attribute

Return type:

dict

execute_on_update()

Execute the on update function

Override this method to provide custom behavior when the object is saved to DynamoDB

Return type:

None

update(**kwargs)

Update the attributes of the object and provide a list of attribute names that were updated.

Keyword arguments: kwargs – Attributes to update

Return type:

list[str]

to_dict(exclude_attribute_names=None, json_compatible=False)

Convert the object to a dict representation

Keyword Arguments: exclude_attribute_names – List of attribute names to exclude from the resulting dict json_compatible – Convert datetime objects to strings and sets to lists for JSON compatibility

Return type:

dict

to_dynamodb_item()

Convert the object to a DynamoDB item

Return type:

dict

Returns:

Dict

to_json()

Convert the object to a JSON string

Return type:

str

Returns:

str

classmethod all_attributes()

Class method that returns all defined attributes on the class

Return type:

list[TableObjectAttribute]

classmethod attribute_definition(name)

Get an attribute definition by name

Keyword Arguments:

attribute (name -- Name of the)

Return type:

TableObjectAttribute | None

Returns:

TableObjectAttribute

classmethod from_dynamodb_item(item)

Create a TableObject from a DynamoDB item

Keyword Arguments:

convert (item -- Item to)

Return type:

TableObject

Returns:

TableObject

classmethod gen_dynamodb_key(partition_key_value, sort_key_value=None)

Generate a DynamoDB key

Keyword Arguments:

• value (sort_key_value -- Sort key)

• value

Return type:

dict

Returns:

Dict

classmethod schema_description()

Get the schema for the object in a human readable format

Return type:

str

Returns:

str

classmethod schema_to_str(only_indexed_attributes=True)

Describe the full schema for the object

Keyword Arguments:

attributes (only_indexed_attributes -- Only describe indexed)

Return type:

str

Returns:

str

static update_date_attributes(date_attribute_names, obj, to_datetime=None)

Update the record_last_updated attribute on the object. Helper method that is commonly used to construct execute_on_update functions.

Keyword Arguments:

• attributes (date_attribute_names -- Names of the date)

• update (obj -- Object to)

• datetime.now() (to_datetime -- Datetime to set, defaults to)

Return type:

None

Table Client

class da_vinci.core.orm.client.TableResultSortOrder(*values)

Bases: StrEnum

Sort order for table query results

ASCENDING = 'ascending'

DESCENDING = 'descending'

class da_vinci.core.orm.client.PaginatorCall

Bases: object

Paginator call types for DynamoDB operations

QUERY = 'query'

SCAN = 'scan'

class da_vinci.core.orm.client.PaginatedResults(items, last_evaluated_key=None)

Bases: object

Container for paginated DynamoDB query results

Keyword Arguments: items – List of table objects returned from the query last_evaluated_key – Key to use for fetching the next page (optional)

__init__(items, last_evaluated_key=None)

class da_vinci.core.orm.client.TableScanDefinition(table_object_class, attribute_prefix=None)

Bases: object

Define filters for scanning a DynamoDB table

Provides a builder pattern for constructing filter expressions for DynamoDB scan and query operations. Supports various comparison operators and handles attribute type conversions.

__init__(table_object_class, attribute_prefix=None)

Create a new scan definition for a DynamoDB table

Keyword Arguments:

• table_object_class – Table object class to scan

• attribute_prefix – Prefix to use for attribute names (default: None)

add(attribute_name, comparison, value)

Add an attribute filter to the scan definition

Keyword Arguments:

• attribute_name – Name of the attribute to filter on

• comparison – Comparison operator to use (ex: equal or greater_than)

• value – Value to compare against

Return type:

None

to_expression()

Convert the scan definition to a DynamoDB expression

Return type:

tuple[str, dict[str, Any]] | tuple[None, None]

Returns:

DynamoDB expression

to_instructions()

Convert the scan definition to a list of basic scan instructions

Return type:

list[str]

Returns:

List of DynamoDB scan instructions

__getattr__(attr)

Override to dynamically add attribute filters to the scan definition using comparison names as method names.

Example: scan_definition.equal(‘foo’, ‘bar’) will add a filter for the attribute foo equal to bar

Keyword Arguments:

attr – Name of the attribute to filter on

Return type:

Any

class da_vinci.core.orm.client.TableClient(default_object_class, app_name=None, deployment_id=None, table_endpoint_name=None, resource_discovery_storage_solution=None)

Bases: object

Client for interacting with DynamoDB tables using Da Vinci table objects

Provides high-level operations for querying, scanning, and managing DynamoDB tables with automatic resource discovery and pagination support.

__init__(default_object_class, app_name=None, deployment_id=None, table_endpoint_name=None, resource_discovery_storage_solution=None)

Initialize the table client with resource discovery

Keyword Arguments: default_object_class – TableObject subclass defining the table schema app_name – Application name for resource discovery (optional) deployment_id – Deployment identifier for resource discovery (optional) table_endpoint_name – Explicit table name (optional, will be discovered if not provided) resource_discovery_storage_solution – Storage solution for resource discovery (optional)

classmethod table_resource_exists(table_object_class, app_name=None, deployment_id=None)

Check if a DaVinci based DynamoDB table exists

Parameters:

• app_name (str | None) – Name of the application

• deployment_id (str | None) – Unique identifier for the installation

• table_object_class (type[TableObject]) – The object class of the table to check

Return type:

bool

paginated(call='query', last_evaluated_key=None, last_evaluated_object=None, limit=None, max_pages=None, parameters=None, sort_order=TableResultSortOrder.ASCENDING)

Handle paginated DynamoDB table results. The last item in a page should be the last evaluated item.

Keyword Arguments:

• call – Name of the DynamoDB client method to call, either a scan or query (default: query)

• last_evaluated_key – Last evaluated key from a previous page of results (default: None)

• last_evaluated_object – Last evaluated object from a previous page of results (default: None), only supported for query

• limit – Maximum number of items to retrieve per page (default: None)

• max_pages – Maximum number of pages to retrieve, if None it will return all available (default: None)

• parameters – Parameters to pass to the client method

• sort_order – Sort order to use for the results, only works for query calls (default: ASCENDING)

Return type:

Generator[PaginatedResults, None, None]

get_object(partition_key_value, sort_key_value=None, consistent_read=False)

Retrieve a single object from the table by partition and sort key

Keyword Arguments:

• partition_key_value – Value of the partition key

• sort_key_value – Value of the sort key (default: None)

• consistent_read – Whether to use consistent read (default: False)

Return type:

TableObject | None

put_object(table_object)

Save a single object to the table

Keyword Arguments:

table_object – Object to save

Return type:

None

delete_object_by_key(partition_key_value, sort_key_value=None)

Delete a single object from the table by partition and sort key

Keyword Arguments:

• partition_key_value – Value of the partition key

• sort_key_value – Value of the sort key (default: None)

Return type:

None

delete_object(table_object)

Delete a single object from the table

Keyword Arguments:

table_object – Object to remove

Return type:

None

scanner(scan_definition)

Perform a scan on the table, works similar to the paginator.

Keyword Arguments:

scan_definition – Scan definition to use (default: None)

Return type:

Generator[PaginatedResults, None, None]

full_scan(scan_definition)

Perform a full scan on the table, returns all items matching the scan definition at once.

Keyword Arguments:

scan_definition – Scan definition to use (default: None)

Return type:

list[TableObject]

update_object(partition_key_value, sort_key_value, updates=None, remove_keys=None)

Updates an item in the DynamoDB table by applying SET and REMOVE operations.

This method allows partial updates to items by setting new values for attributes or removing existing ones. It supports dot notation for nested JSON map updates, enabling the modification of specific keys within a JSON-like structure in DynamoDB.

Parameters:

• partition_key_value (Any) – The value of the partition key for the item to be updated.

• sort_key_value (Any) – The value of the sort key for the item to be updated.

• updates (Dict[str, Any], optional) – A dictionary containing attribute names (as keys) and their new values (as values) to be updated in the table. If dot notation is used in the attribute name (e.g., ‘json_map.sub_key’), it will update a nested key within a DynamoDB MAP type.

• remove_keys (List[str], optional) – A list of attribute names to be removed from the item. Dot notation can be used to remove nested attributes from a DynamoDB MAP.

Example Usage:

• To update a nested key inside a JSON map:

  updates = {‘json_map.sub_key’: ‘new_value’} remove_keys = [‘json_map.another_sub_key’]

• To update a top-level attribute and remove another:

  updates = {‘attribute1’: ‘new_value’} remove_keys = [‘attribute2’]

Notes

• This method generates DynamoDB UpdateExpressions to execute the SET and REMOVE

operations in a single request. - If both updates and remove_keys are provided, they are combined in the final update expression. - Dot notation in updates or remove_keys will handle nested attributes within DynamoDB MAP types. - This method assumes the object’s table schema is already defined in the default_object_class.

Raises:

• ClientError – If a client error occurs during the DynamoDB update operation.

• Exception – If an attribute with an empty value is provided for a DynamoDB JSON attribute.

Return type:

None

Returns:

None

ORM Exceptions

Custom Exceptions for the ORM

exception da_vinci.core.orm.orm_exceptions.AugmentedRetrievalInvalidQueryError(query, details)

Bases: ValueError

__init__(query, details)

Indicates that an invalid query was generated for an augmented retrieval

Keyword Arguments:

• query (str) – The invalid query

• details (str) – Details about why the query is invalid

exception da_vinci.core.orm.orm_exceptions.MissingTableObjectAttributeError(attribute_name)

Bases: ValueError

__init__(attribute_name)

Indicates that a required attribute was not provided to a TableObject

Keyword Arguments:

attribute_name (str) – The name of the attribute that was not provided

exception da_vinci.core.orm.orm_exceptions.TableScanQueryError(attribute_name, attribute_type)

Bases: ValueError

__init__(attribute_name, attribute_type)

Indicates that an invalid attribute_name was provided for a given attribute_type

Keyword Arguments:

• attribute_name (str) – The name of the attribute that was invalid

• attribute_type (str) – The type of attribute that was invalid

exception da_vinci.core.orm.orm_exceptions.TableScanInvalidComparisonError(comparison_name)

Bases: TableScanQueryError

__init__(comparison_name)

Raised when an invalid comparison operator was provided

Keyword Arguments:

comparison_name (str) – The name of the invalid comparison operator

exception da_vinci.core.orm.orm_exceptions.TableScanInvalidAttributeError(attribute_name)

Bases: TableScanQueryError

__init__(attribute_name)

Raised when attempting to scan using an invalid attribute

Keyword Arguments:

attribute_name (str) – The name of the invalid attribute

exception da_vinci.core.orm.orm_exceptions.TableScanMissingAttributeError(attribute_name)

Bases: ValueError

__init__(attribute_name)

Used when a attribute is missing from a table scan

Keyword Arguments:

attribute_name (str) – The name of the missing attribute

Core Module

Exceptions

All Da Vinci Base Exceptions

exception da_vinci.core.exceptions.DuplicateRouteDefinitionError(route_name)

Bases: Exception

__init__(route_name)

Base error for the Da Vinci framework REST Service Base class

Indicates that a route definition already exists

Keyword Arguments:

route_name (str) – The name of the route that already exists

exception da_vinci.core.exceptions.GlobalSettingsNotEnabledError

Bases: Exception

__init__()

Indicates that global settings are not enabled

exception da_vinci.core.exceptions.GlobalSettingNotFoundError(setting_key, namespace)

Bases: Exception

__init__(setting_key, namespace)

Indicates that a global setting was not found

Keyword Arguments:

• setting_key (str) – The key of the setting that was not found

• namespace (str) – The namespace of the setting that was not found

exception da_vinci.core.exceptions.MissingRequiredRuntimeVariableError(variable_name)

Bases: RuntimeError

__init__(variable_name)

Indicates that a required runtime variable used by the Da Vinci framework was not found

Keyword Arguments:

variable_name (str) – The name of the variable that was not found

exception da_vinci.core.exceptions.ResourceNotFoundError(resource_name, resource_type)

Bases: Exception

__init__(resource_name, resource_type)

Resource was not able to be located using Da Vinci resource discovery

Keyword Arguments:

• resource_name (str) – The name of the resource that was not found

• resource_type (str) – The type of resource that was not found

Resource Discovery

Resource Discovery Module

class da_vinci.core.resource_discovery.ResourceType(*values)

Bases: StrEnum

Resource types registered with service discovery

ASYNC_SERVICE = 'async_service'

BUCKET = 'bucket'

DOMAIN = 'domain'

REST_SERVICE = 'rest_service'

TABLE = 'table'

class da_vinci.core.resource_discovery.ResourceDiscoveryStorageSolution(*values)

Bases: StrEnum

Resource discovery storage solutions

DYNAMODB = 'dynamodb'

SSM = 'ssm'

class da_vinci.core.resource_discovery.ResourceDiscovery(resource_type, resource_name, app_name=None, deployment_id=None, storage_solution=None)

Bases: object

__init__(resource_type, resource_name, app_name=None, deployment_id=None, storage_solution=None)

Initialize a ResourceDiscovery object

Keyword Arguments:

• resource_type – Type of the resource

• resource_name – Name of the resource

• app_name – Name of the application (default: None)

• deployment_id – Unique identifier for the installation (default: None)

classmethod ssm_parameter_name(resource_type, resource_name, app_name, deployment_id)

Return the parameter for a resource registered with service discovery.

Keyword Arguments:

• resource_type – Type of the resource

• resource_name – Name of the resource

• app_name – Name of the application

• deployment_id – Unique identifier for the installation

Return type:

str

endpoint_lookup()

Return the endpoint for a resource registered with service discovery.

Return type:

str

service_registry_table_lookup()

Return the endpoint for a resource registered with service discovery using the service registry table.

Return type:

str

ssm_resource_endpoint_lookup()

Return the endpoint for a resource registered with service discovery. The app_name and deployment_id parameters are optional. If not provided, the values will be loaded from the environment.

Keyword Arguments:

• resource_type – Type of the resource

• resource_name – Name of the resource

• app_name – Name of the application (default: None)

• deployment_id – Unique identifier for the installation (default: None)

Return type:

str

Global Settings

da_vinci.core.global_settings.global_settings_available()

Check if global settings are available

Try the environment variable first, then check if the settings table exists if the environment variable is not set.

Return type:

bool

da_vinci.core.global_settings.setting_value(namespace, setting_key)

Retrieve a setting value as the correct Python type, given a namespace and key

Parameters:

• setting_key (str) – The setting key

• namespace (str) – The namespace of the setting

Return type:

Any | None

Logging

class da_vinci.core.logging.S3LogHandler(execution_id, namespace, metadata=None)

Bases: Handler

Custom log handler to store logs in memory and offload them to S3.

__init__(execution_id, namespace, metadata=None)

Set up the S3 log handler with the execution ID and namespace.

Keyword Arguments: execution_id – The execution ID namespace – The namespace for the logs metadata – Additional metadata to store with the logs

emit(record)

Capture log records and store them in memory.

Keyword Arguments: record – The log record

Return type:

None

get_log_entries()

Return the collected log entries.

Return type:

list

put_metadata(key, value)

Add metadata to the log handler.

Keyword Arguments: key – The key for the metadata value – The value for the metadata

Return type:

None

to_dict()

Return the collected log entries as a dictionary.

Return type:

dict

class da_vinci.core.logging.Logger(namespace, log_level_name=None, s3_logging_enabled=False, s3_logging_bucket_name=None)

Bases: object

__init__(namespace, log_level_name=None, s3_logging_enabled=False, s3_logging_bucket_name=None)

Initialize the logger with the namespace and log level name.

Keyword Arguments: namespace – The namespace for the logger log_level_name – The log level name (default: INFO) s3_logging_enabled – Enable logging to S3 (default: False) s3_logging_bucket_name – The S3 bucket name for logging (default: None)

dump_to_s3()

Dump the collected logs to an S3 bucket as a JSON file.

Return type:

None

finalize()

Call this method to finalize logging and upload logs to S3 if enabled.

Return type:

None

debug(message)

Add a log message with level DEBUG.

Keyword Arguments: message – The log message

Return type:

None

info(message)

Add a log message with level INFO.

Keyword Arguments: message – The log message

Return type:

None

warning(message)

Add a log message with level WARNING.

Keyword Arguments: message – The log message

Return type:

None

error(message)

Add a log message with level ERROR.

Keyword Arguments: message – The log message

Return type:

None

JSON Utilities

JSON Utility functionality

class da_vinci.core.json.DateTimeEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

JSONEncoder class that encodes datetime objects strings

default(obj)

Encode datetime objects as ISO format strings

Keyword Arguments: obj – Object to encode

Raises:

TypeError -- If object is not JSON serializable –

Return type:

Any

class da_vinci.core.json.DaVinciObjectEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

JSONEncoder class that encodes commonly used framework objects

default(obj)

Encode Da Vinci framework objects and datetime objects

Handles ObjectBody instances by converting to dictionaries and datetime objects by converting to ISO format strings.

Keyword Arguments: obj – Object to encode

Raises:

TypeError -- If object is not JSON serializable –

Return type:

Any

Event Bus Module

Event Publisher

Exception Trap Module