How to create a class for a Python Table

Steps#

In the portal go to settings and click on Brick Management (v2).
Click the Create Brick button in the topright corner of the screen, or select an existing brick from the list when you need to extend or update an existing brick.

When updating a brick you can choose to make a Fork from an existing brick and build a new brick in the copied fork, or Create Version which updates the brick to a new version.

Go to the code tab and select the file you wish to update or create a new one.
Create a python class in this file.

Creating the class examples#

The class needs a query method like this:

def query(self, filters, *args, **kwargs):

In this method get the data you wish to create a table from and make sure to return it via:

rb = pa.RecordBatch.from_pandas(df_2)
yield rb

Where df_2 is the data you wish to create the table from. If there are filters make sure to apply them now, these are created from the WHERE clause in any query.

  def repl(self, match):
      str_to_replace = match[0]
      if len(str_to_replace) % 2 == 0:
          return str_to_replace
      return str_to_replace.replace("%", "*").replace("^", "_")
  def recurse_filter(self, _filter):
      if (type(_filter.left) == rivendel.BinaryExpr) & (
          type(_filter.right) == rivendel.BinaryExpr
      ):
          name, left_values = recurse(_filter.left)
          name, right_values = recurse(_filter.right)
          return name, left_values + right_values

      elif (type(_filter.left) == rivendel.BinaryExpr) & (
          type(_filter.right) != rivendel.BinaryExpr
      ):

          name, left_values = recurse(_filter.left)
          right_value = _filter.right.value
          return name, left_values + [right_values]

      elif (type(_filter.left) != rivendel.BinaryExpr) & (
          type(_filter.right) == rivendel.BinaryExpr
      ):

          left_value = _filter.left.value
          name, right_values = recurse(_filter.right)
          return name, [left_value] + right_values
        
      else:
          name = _filter.left.name
          right_value = _filter.right.value
          return name, [right_value]

  def get_instance_filters(self, filters):
      instance_filters = {}
      for _filter in filters:
          if isinstance(_filter, rivendel.BinaryExpr):
              if _filter.operator == "LIKE":
                  name = _filter.left.name
                  vals = [re.sub(r"([\\]*\^)|([\\]*\%)", self.repl, _filter.right.value)]
              else:
                  name, vals = self.recurse_filter(_filter)  
              if name in instance_filters.keys():
                  instance_filters[name] += vals
              else:
                  instance_filters.update({
                      name: vals
                  })
        
      return instance_filters

  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,
          )
      )

These methods combined will change the filters into a usable form.

If your data is a pandas DataFrame, you can now filter it easily with

  for key, values in instance_filters.items():
      if key in df.columns:
          df = df.loc[df[key].isin(values), :]

Alternatively you can write your own custom way of filtering your data using the filters.

Using secrets#

If you wish to use secrets you can retrieve them by importing rivendel, and assigning json.loads(rivendel.Secrets.get('brick-secrets', secret_path)) to a variable.