This data layer also supports the BaseStorageClient that enables you to store your elements into AWS S3 or Azure Blob Storage.

​
Example

Here is an example of setting up this data layer. First install boto3:

pip install boto3

Import the custom data layer and storage client, and set the cl_data._data_layer variable at the beginning of your Chainlit app.

import chainlit.data as cl_data
from chainlit.data.dynamodb import DynamoDBDataLayer
from chainlit.data.storage_clients import S3StorageClient

storage_client = S3StorageClient(bucket="<Your Bucket>")

cl_data._data_layer = DynamoDBDataLayer(table_name="<Your Table>", storage_provider=storage_client)

​
Table structure

Here is the Cloudformation used to create the dynamo table:

{
  "AWSTemplateFormatVersion": "2010-09-09",
  "Resources": {
    "DynamoDBTable": {
      "Type": "AWS::DynamoDB::Table",
      "Properties": {
        "TableName": "<YOUR-TABLE-NAME>",
        "AttributeDefinitions": [
          {
            "AttributeName": "PK",
            "AttributeType": "S"
          },
          {
            "AttributeName": "SK",
            "AttributeType": "S"
          },
          {
            "AttributeName": "UserThreadPK",
            "AttributeType": "S"
          },
          {
            "AttributeName": "UserThreadSK",
            "AttributeType": "S"
          }
        ],
        "KeySchema": [
          {
            "AttributeName": "PK",
            "KeyType": "HASH"
          },
          {
            "AttributeName": "SK",
            "KeyType": "RANGE"
          }
        ],
        "GlobalSecondaryIndexes": [
          {
            "IndexName": "UserThread",
            "KeySchema": [
              {
                "AttributeName": "UserThreadPK",
                "KeyType": "HASH"
              },
              {
                "AttributeName": "UserThreadSK",
                "KeyType": "RANGE"
              }
            ],
            "Projection": {
              "ProjectionType": "INCLUDE",
              "NonKeyAttributes": ["id", "name"]
            }
          }
        ],
        "BillingMode": "PAY_PER_REQUEST"
      }
    }
  }
}

​
Logging

DynamoDB data layer defines a child of chainlit logger.

import logging
from chainlit import logger

logger.getChild("DynamoDB").setLevel(logging.DEBUG)

​
Limitations

Filtering by positive/negative feedback is not supported.

The data layer methods are not async. Boto3 is not async and therefore the data layer uses non-async blocking io.

​
Design

This implementation uses Single Table Design. There are 4 different entity types in one table identified by the prefixes in PK & SK.

Here are the entity types:

type User = {
    PK: "USER#{user.identifier}"
    SK: "USER"
    // ...PersistedUser
}

type Thread = {
    PK: f"THREAD#{thread_id}"
    SK: "THREAD"
    // GSI: UserThread for querying in list_threads
    UserThreadPK: f"USER#{user_id}"
    UserThreadSK: f"TS#{ts}"
    // ...ThreadDict
}

type Step = {
    PK: f"THREAD#{threadId}"
    SK: f"STEP#{stepId}"
    // ...StepDict

    // feedback is stored as part of step. 
    // NOTE: feedback.value is stored as Decimal in dynamo which is not json serializable
    feedback?: Feedback
}

type Element = {
    "PK": f"THREAD#{threadId}"
    "SK": f"ELEMENT#{element.id}"
    // ...ElementDict
}

Was this page helpful?

Suggest editsRaise issue
SQLAlchemy Data LayerTags & Metadata