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.