Fusion Mesh Bricks

Purpose	Explore this guide to gain a deeper understanding of Fusion Mesh Bricks and their setup process. We will walk you through various elements you can use in Python and provide step-by-step instructions for configuring them.

Last Updated	May 19, 2024

---

Mesh Bricks enable the translation of queries from the Raven query app to various external data sources, such as web APIs, (No)SQL databases, or crawled web pages.

About Tables

In the Raven Portal, different Tables are available, each requiring specific configurations. This document will focus on the Mesh Table and its corresponding Brick.

Bricks within tables are responsible for holding information and managing the table's configuration, including how data is retrieved and structured.

To learn more about tables, follow the link below:

Learn more about Tables

Creating a Fusion Mesh Brick

To create a new Fusion Mesh Brick:

1. Navigate to Settings / Brick Management V3 and click CREATE BRICK in the top right corner.

2. Fill in the name and description, and select Fusion Mesh as the type.

If you have an existing Fusion Mesh Brick, you can use the fork option to skip some configuration steps.

Adding Parameters

To add parameters to your brick:

1. Navigate to Configuration / Parameters.

2. Click ADD PARAMETER.

3. Configure the parameters using the following options:

Name	Description
Name	Parameter key (params['key'])
Title	Parameter name in table configuration
Description	Parameter description in table configuration
Type	Parameter type (String, Number, Secret)
Default	Default fallback value for when parameter is not configured

Required Parameters

The following parameters are essential for Fusion Mesh Bricks:

Name	Type	Description
main	String	Name of file in which the main brick code is located
class_name	String	Name of brick class

Writing the Brick Code

To start coding your brick:

1. Navigate to Code.

2. Click + to add a Python file.

Ensure the file name matches the value configured in the main parameter.

Brick Class

The Brick class name should match the value in the class_name parameter. This class must implement the following methods:

__init__

This method initializes the brick when a query is executed.

Parameters:

params (dict): A dictionary of parameter values configured during setup.

Returns: None

query

This method is called with a query's filters when a query runs. The engine will apply filters, sorting, grouping, and limiting to the results.

Parameters:

• filters (List): Filters specified in the query.

• sort (Object): Sorting information.

  • sort.expr.name (String): Column name for sorting.
  • sort.asc (Boolean): Sort order (ascending or descending).

• limit (Int): Maximum number of rows to return.

Returns: data (pyarrow.RecordBatch)

scan

Typically, this method can be implemented as follows for most use cases:

def scan(self):
    return PyStatistics(
        num_rows=None,
        is_exact=False,
        total_byte_size=None,
        column_statistics=PyColumnStatistics(
            null_count=None,
            distinct_count=None,
            max_value=None,
            min_value=None,
        )
    )

Handling Filters

Filters can be complex. Here’s a code snippet to translate filters to a SQL WHERE statement:

def generate_where(self, filters):
    def to_sql(expr):
        try:
            if isinstance(expr, rivendel.Literal):
                if isinstance(expr.value, str):
                    return f"'{expr.value}'"
                elif isinstance(expr.value, (int, float)):
                    return f"{expr.value}"
                elif isinstance(expr.value, bool):
                    return 'true' if expr.value else 'false'
                elif isinstance(expr.value, datetime.datetime):
                    return f"'{expr.value.strftime('%Y-%m-%d %H:%M:%S.%f')}'"
                else:
                    return f"{expr.value}"
            elif isinstance(expr, rivendel.Column):
                return f"{expr.name}"
            elif isinstance(expr, rivendel.Like):
                return f"{to_sql(expr.expr)} {'NOT ' if expr.negated else ''}LIKE {to_sql(expr.pattern)}"
            elif isinstance(expr, rivendel.Alias):
                return to_sql(expr.expr)
            elif isinstance(expr, rivendel.IndexedField):
                return f"{to_sql(expr.expr)}.{expr.key}"
            elif isinstance(expr, rivendel.Cast):
                return to_sql(expr.expr)
            elif isinstance(expr, rivendel.ScalarVariable):
                return str(expr)
            elif isinstance(expr, str):
                return f"'{expr}'"
            elif isinstance(expr, rivendel.IsNull):
                return f"{to_sql(expr.expr)} IS NULL"
            elif isinstance(expr, rivendel.IsNotNull):
                return f"{to_sql(expr.expr)} IS NOT NULL"
            elif isinstance(expr, rivendel.InList):
                in_list = ', '.join(to_sql(item) for item in expr.list)
                return f"{to_sql(expr.expr)} {'NOT ' if expr.negated else ''}IN ({in_list})"
            elif isinstance(expr, rivendel.BinaryExpr):
                return f"({to_sql(expr.left)} {expr.operator} {to_sql(expr.right)})"
            elif isinstance(expr, rivendel.IsFalse):
                return f"({to_sql(expr.expr)} = false)"
            elif isinstance(expr, rivendel.IsTrue):
                return f"({to_sql(expr.expr)} = true)"
            return f"NOT SUPPORTED YET({expr} {type(expr)})"
        except:
            return None

    where_clauses = [to_sql(f) for f in filters if to_sql(f) is not None]
    return " WHERE " + ' AND '.join(where_clauses) if where_clauses else ""

Handling Secrets

To retrieve secrets from a secret parameter named credentials, use the following code:

credentials = params.get('credentials')
secret = Secrets.get("brick-secrets", credentials)
import json
secret = json.loads(secret)

The secret variable will now contain a dictionary with keys based on the supplied secret type.

For more information on secret types and setup, follow the link below:

Learn more about Secrets