Skip to main content

luna.connections

ConnectionType Objects​

class ConnectionType(str, enum.Enum)

The type of data connection.

ConnectionValueType Objects​

class ConnectionValueType(str, enum.Enum)

The type of value coming from a data connection.

Connection targets will produce values of a specific type. Operations in connection mappings will then transform the data, each one having an input and output type that allows us to verify chains of operations are valid and know what type they ultimately produce.

These types exist so that we can have a clear, unambiguous declaration of the types that can be handled by connection mappings, and enable us to validate operations. However, we don't want to expose these types to users, and for actual contract connections you still use actual Python types (which can then interact with the Luna types when going back and forth with the user interface).

HttpMethod Objects​

class HttpMethod(str, enum.Enum)

The HTTP method used in an HTTP connection.

Table Objects​

class Table(BaseModel)

Simple abstraction to represent data tables.

Rows are 1-indexed, as this is more familiar to users.

We suspect a lot of people who want to inject tables into their contracts and work with them will want to use Pandas. However, we don't want to force the overhead of Pandas being installed in every runner (it takes a lot of space when you include its dependencies). So, we built this lightweight abstraction to wrap table data, which can easily be converted into a Pandas data frame if someone wants to install Pandas and do that:

>>> table = Table(columns=["A", "B"], data=[[1, True], [2, False]]) >>> df = pandas.DataFrame(columns=table.columns, data=table.data)

get_value​

def get_value(column_name: str, row_index: int) -> Any

Returns the value in a given column and row.

Raises: ValueError if column name invalid. IndexError if row index invalid.

get_column_letter​

def get_column_letter(column_index: int) -> str

Return the alphabetical column index for a given index number.

This follows conventions used by Excel, where the first column (column 1) is 'A', through to 'Z', 'AA', 'AB' etc.

Sheet Objects​

class Sheet(Table)

Simple abstraction to represent data sheets.

This is similar to a table, except instead of column names you have Excel style column letters.

get_table​

def get_table(start_column_letter: str, start_row_index: int,
              end_column_letter: str, end_row_index: int) -> Table

Returns a table from a sub-set of the sheet.

Raises: ValueError if a column name is invalid. IndexError if a row index is invalid.

find_placeholders​

def find_placeholders(value: Any) -> set[str]

Finds placeholder values.

This can be called on a string, list or dictionary, and it will go and find placeholders recursively.

replace_placeholders​

def replace_placeholders(value: Any, replacements: dict) -> Any

Replaces placeholder values.

This can be called on a string, list or dictionary, and it will go and find placeholders recursively.

ConnectionSourceProperties Objects​

class ConnectionSourceProperties(BaseModel)

Base class for properties describing connection sources.

get_required_type​

@classmethod
@abc.abstractmethod
def get_required_type(cls) -> ConnectionType

The type required for a connection source.

Each subclass must declare what their type is. This enables us to parse arbitrary connection source data using the correct type, and to validate that connection source data is given the correct type.

get_placeholders​

def get_placeholders() -> set[str]

Returns placeholders found in the source properties.

ConnectionTargetProperties Objects​

class ConnectionTargetProperties(BaseModel)

Base class for properties describing connection targets.

get_required_type​

@classmethod
@abc.abstractmethod
def get_required_type(cls) -> ConnectionType

The type required for a connection target.

Each subclass must declare what their type is. This enables us to parse arbitrary connection target data using the correct type, and to validate that connection target data is given the correct type.

get_output_type​

@abc.abstractmethod
def get_output_type() -> ConnectionValueType

The type output by the connection target.

get_placeholders​

def get_placeholders() -> set[str]

Returns placeholders found in the target properties.

HttpApiConnectionSourceProperties Objects​

class HttpApiConnectionSourceProperties(ConnectionSourceProperties)

Connection source properties for HTTP API connections.

HttpApiConnectionTargetProperties Objects​

class HttpApiConnectionTargetProperties(ConnectionTargetProperties)

Connection target properties for HTTP API connections.

OdbcDatabaseConnectionSourceProperties Objects​

class OdbcDatabaseConnectionSourceProperties(ConnectionSourceProperties)

Connection source properties for ODBC database connections.

OdbcDatabaseConnectionTargetProperties Objects​

class OdbcDatabaseConnectionTargetProperties(ConnectionTargetProperties)

Connection target properties for ODBC database connections.

S3BucketConnectionSourceProperties Objects​

class S3BucketConnectionSourceProperties(ConnectionSourceProperties)

Connection source properties for S3 buckets.

S3BucketConnectionTargetProperties Objects​

class S3BucketConnectionTargetProperties(ConnectionTargetProperties)

Connection target properties for S3 buckets.

UserEnteredValueConnectionSourceProperties Objects​

class UserEnteredValueConnectionSourceProperties(ConnectionSourceProperties)

Connection source properties for user entered values.

UserEnteredValueConnectionTargetProperties Objects​

class UserEnteredValueConnectionTargetProperties(ConnectionTargetProperties)

Connection target properties for user entered values.

ConnectionMappingOperation Objects​

class ConnectionMappingOperation(BaseModel, abc.ABC)

Data operation applied to a connection mapping.

A connection mapping transforms data coming from a connection target into a value using a series of these operations.

get_input_type​

@classmethod
@abc.abstractmethod
def get_input_type(cls) -> ConnectionValueType

The type the operation expects as input.

get_output_type​

@classmethod
@abc.abstractmethod
def get_output_type(cls) -> ConnectionValueType

The type the operation expects as output.

execute​

@abc.abstractmethod
def execute(input_value: Any) -> Any

Perform the operation and return an output value.

get_placeholders​

def get_placeholders() -> set[str]

Returns placeholders found in the operation properties.

SelectSheetFromXlsxWorkbook Objects​

class SelectSheetFromXlsxWorkbook(ConnectionMappingOperation)

Select a sheet of data from an Excel .xlsx workbook.

ParseSheetFromString Objects​

class ParseSheetFromString(ConnectionMappingOperation)

Parse a sheet of data from a string of text (e.g. CSV file data).

SelectTableFromSheet Objects​

class SelectTableFromSheet(ConnectionMappingOperation)

Select a table of data from a sheet

SelectValueFromJson Objects​

class SelectValueFromJson(ConnectionMappingOperation)

Select a single value from JSON data using JSON path.

execute​

def execute(input_value: Any) -> Any

Finds values in JSON data using the JSON path.

If there is a single match, then that value is returned (this might be any valid JSON, including a single value, an array (as a list) or an object (as a dictionary). If there are multiple matches, we return an array (as a list) of all matches.

This function accepts JSON types, meaning dictionaries, lists or single values. If a string is passed in, it is interpreted as a single string. If a connection mapping has a string containing JSON data which they want to perform this operation on, they must first use another operation (e.g. ParseValueAsJson) to convert it to JSON.

SelectValueFromTableByIndex Objects​

class SelectValueFromTableByIndex(ConnectionMappingOperation)

Get a value using column name and row index.

Rows are 1-indexed, as this is more familiar to users.

ParseValueAsString Objects​

class ParseValueAsString(ConnectionMappingOperation)

Parse an untyped value into a string.

execute​

def execute(input_value: Any) -> str | None

Parses the value as a string.

There is no error handling because this should always work in Python unless something crazy has happened (like a bug in Python), or someone has deliberately made str raise an error.

ParseValueAsDecimal Objects​

class ParseValueAsDecimal(ConnectionMappingOperation)

Parse an untyped value into a decimal.

If decimal places are specified, then we quantize the number to match that. If they are not specified, then we will use the maximum available. Numbers are rounded half up.

execute​

def execute(input_value: Any) -> Decimal | None

Parses the value as a decimal.

ParseValueAsInteger Objects​

class ParseValueAsInteger(ConnectionMappingOperation)

Parse an untyped value into an integer.

execute​

def execute(input_value: Any) -> int | None

Parses the value as an integer.

ParseValueAsDatetime Objects​

class ParseValueAsDatetime(ConnectionMappingOperation)

Parse an untyped value into a datetime.

execute​

def execute(input_value: Any) -> datetime | None

Parses the value as a datetime.

ParseValueAsDate Objects​

class ParseValueAsDate(ConnectionMappingOperation)

Parse an untyped value into a date.

execute​

def execute(input_value: Any) -> date | None

Parses the value as a datetime.

ParseValueAsTime Objects​

class ParseValueAsTime(ConnectionMappingOperation)

Parse an untyped value into a time.

execute​

def execute(input_value: Any) -> time | None

Parses the value as a time.

ParseValueAsTimeDelta Objects​

class ParseValueAsTimeDelta(ConnectionMappingOperation)

Parse an untyped value into a time delta.

execute​

def execute(input_value: Any) -> timedelta | None

Parses the value as a time delta.

ParseValueAsJson Objects​

class ParseValueAsJson(ConnectionMappingOperation)

Parse an untyped value into JSON.

execute​

def execute(input_value: Any) -> Any

Parses the value as JSON data.

get_connection_value_type​

def get_connection_value_type(type_name: str) -> ConnectionValueType

Get the connection value type needed for values of a given type.

Any connection in a contract will expect values of a certain type to be produced by mappings used for that connection. This function tells you what type the mapping must output to produce the required type.

This accepts strings as inputs to make it easier to work with types which are stored as part of template info.

parse_operations​

def parse_operations(
    operations_properties: list[dict[str, Any]]
) -> list[ConnectionMappingOperation]

Parses operations data into a list of operations objects.

RuntimeConnection Objects​

class RuntimeConnection(BaseModel)

A connection that can be used to read data.

A runtime connection is created by taking data which describes the properties of the connection, and using these to build an actual connection which can read data at runtime.

See RuntimeConnectionFactory.

build​

@classmethod
@abc.abstractmethod
def build(cls, name: str, source_properties: dict[str, Any],
          target_properties: dict[str, Any],
          operations_properties: list[dict[str, Any]]) -> "RuntimeConnection"

Build the object from data passed into the kernel.

read_raw​

@abc.abstractmethod
def read_raw() -> Any

Return data direct from the connection target.

read​

def read() -> Any

Return data after processing using the mapping operations.

HttpApiConnection Objects​

class HttpApiConnection(RuntimeConnection)

A runtime connection for HTTP APIs.

get_url​

def get_url() -> str

Builds the full URL to call to get data from this connection.

get_request_headers​

def get_request_headers() -> dict[str, str]

Builds the full set of request headers.

read_raw​

def read_raw() -> Any

Returns JSON data from the API.

OdbcDatabaseConnection Objects​

class OdbcDatabaseConnection(RuntimeConnection)

A runtime connection for ODBC databases.

read_raw​

def read_raw() -> Any

Returns a table of data from the connection.

S3BucketConnection Objects​

class S3BucketConnection(RuntimeConnection)

A Runtime connection for S3 buckets.

read_raw​

def read_raw() -> Any

Returns data from the object in S3.

This will return different types based on the configured output type for the connection target (for example, if it is XLSX data you get an Excel workbook file). If you want to use the connection to access the raw data in the object, use read_data.

read_data​

def read_data() -> bytes

Returns raw data from the object in S3.

UserEnteredValueConnection Objects​

class UserEnteredValueConnection(RuntimeConnection)

A Runtime connection for user entered values.

RuntimeConnectionFactory Objects​

class RuntimeConnectionFactory()

Builds runtime connections using connection data.

build​

@staticmethod
def build(name: str, data: GetConnectionResponse) -> RuntimeConnection

Build and return the runtime connection.