Attributes and Methods

config

Your function will be given your Node's runtime configuration in the config attribute. This attribute is a Python dict.

NOTE - this is the runtime configuration for your Node. It will be the union of your Tenant's configuration and your Node's configuration.

This can be used to get any sort of configuration that you function may require - e.g. database credentials.

For example, using a processor function:

def processor(*, context, message, source, **kwargs):
    
    # Assuming that dbUsername, dbPassword, and dbEndpoint
    # were set in either your Tenant's config or your Node's config!
    db_username = context.config["dbUsername")
    db_password = context.config["dbPassword")
    db_endpoint = context.config["dbEndpoint")
    
    # Establish DB connection here, and augment message

    return message

handle_bulk_data

You may write a function that requires the ability to store bulk data to reduce message size. Normally this will only be useful in processor functions.

The Context object provides you easy access to the Bulk Data Storage capabilities of EchoStream via the handle_bulk_data method. This method has the following signature:

def handle_bulk_data(
    data: Union[bytearray, bytes, BinaryIO], 
    *, 
    contentEncoding: Literal["deflate", "gzip"] = "gzip"
) -> str

• data: the bulk data that you wish to store. May be a bytearray, bytes or a BinaryIO object
• contentEncoding: the compression to apply to the object when it is stored. Either deflate or gzip. Defaults to gzip.

When called, handle_bulk_data will take the data you provide, compress it according to contentEncoding, store it in your Tenant's Bulk Data Storage area, and return to you a URL that allows you to download that data.

For eaxmple, let's assume that during processing you need to call an API, retrieve a large binary object (e.g. - an image), and attach that image to the message that you forward on.

def processor(*, context, message, source, **kwargs):
    
    # message is stringified JSON (example only) so must load it
    message = json.loads(message)
    
    # A bunch of data
    data = b'0' * 1000000
    
    # Put the returned URL in the message as a "ticket"
    message["dataUrl"] = context.handle_bulk_data(data)

    # Return the message stringified
    return json.dumps(message)

logger

Often you will want to log messages from within your function, and then subsequently have those log message made available to you (see Log monitoring).

The Context object provides you the logger attribute to do just that. This attribute is a Python Logger object, and it must be used to log messages from within your function.

For example, using a processor function:

def processor(*, context, message, source, **kwargs):
    
    context.logger.info(f"This is the received message {message}")

    return message

node

The Context object contains the attribute node. This is a string that is the name of the Node that is running your function.

By passing the current Node into your function (via the Context object), EchoStream enables you to write generic code that can be used across Nodes (or placed in the Function Library). Without this, your function would not know what Node it was being executed in.

This is useful if your function needs to know what Node it is operating in, either for logging or conditional processing.

For example, using a bitmapper function:

def bitmapper(*, context, message, source, **kwargs):

    bitmap = 0x0
    
    if "Foo" in context.node:
        bitmap |= 0x1
    if "Bar" in context.node:
        bitmap |= 0x2
    
    context.logger.info(f"Node name: {context.node}\nbitmap: {bitmap}")

    return bitmap

table

As stated in the Tenant section, every Tenant is provided a DynamoDB Table to store inter-Node or inter-execution state into.

The Context object exposes this Table to your function via the table attribute. This attribute will return to you a boto3 Table resource. The returned Table resource is thread-safe.

For example, using a processor function:

def processor(*, context, message, source, **kwargs):
    
    from simplejson import json
    
    # message is stringified JSON (example only) so must load it
    message = json.loads(message)
    
    # Get "foo.bar" from the table
    item = context.table.get_item(Key=dict(pk="foo", sk="bar"))
    # If "foo.bar" exists and has the field "myVal", copy that into the message as "theVal"
    if item and (val := item.get("myVal")):
        message["theVal"] = val

    # Return the message stringified
    return json.dumps(message)

You may use this Table in any way that you wish.

tenant

The Context object contains the attribute tenant. This is a string that is the name of the Tenant that contains the Node that is running your function.

By passing the current Tenant into your function (via the Context object), EchoStream enables you to write generic code that can be used in one or more Tenants. Without this, your function would not know what Tenant it was being executed in.

This is useful if your function needs to know what Tenant it is operating in, either for logging or conditional processing.