How To Guide: Query a PostegreSQL Table in Realtime

Create a simple API Resource that gets information from a PostgreSQL table for use in a Realtime Feature View.

When making real-time predictions, you often need to enrich your features with the latest operational data - like checking a user's current account balance or recent transaction history stored in an operational database such as PostgreSQL - to make accurate decisions. API Resources make this possible by maintaining a fast, persistent connection to your production database, allowing you to seamlessly incorporate this real-time data into your feature computations.

For this example we will query an operational table with bank users data to retrieve features such as phone_number, balance, and credit_score.

Introduction

This guide assumes some basic familiarity with Tecton, including how to create and apply features. You will also need to create Tecton Secrets.

If you are new to Tecton, we recommend first checking out our Tutorials which cover a range of topics designed to bring you up to speed working with Tecton.

While following this guide, you will:

• Create a Tecton Secret to hold your PostgreSQL connection string
• Give your workspace access to the Tecton Secret
• Create a Tecton environment
• Create and Apply a Transform Server Group
• Apply a Resource Provider using the Tecton Secret and a Realtime Feature View (RTFV)
• Apply a Feature Service using the Transform Server Group and include the RTFV
• Run a sample call to retrieve values from the table

Prerequisites

1. A production Tecton workspace (this guide will not work with a lab workspace)
2. A Tecton service account API key
3. A PostgreSQL database and associated connection string, which will look something like: postgresql://<username>:<password>@<hostname>:<port>/<database>?sslmode=require

note

When connecting to operational databases, reach out to Tecton support to configure access to the specified database port.

Step 1: Create a Tecton Secret for Your PostgreSQL Connection String

Tecton Secrets allow you to create, store, and access credentials securely.

1. Create a connection string for your PostgreSQL database, and store it in a file. In our example, it's called api-key.txt
2. Create a Scope and a Secret for the connection string in Tecton:
   1. tecton secrets create-scope -s <scope-name>
   2. tecton secrets put -s <scope-name> -k <key-name> -f api-key.txt
3. Give your workspace access to the above secret scope:
   1. tecton access-control assign-role -w <workspace> -r secret_scope_reader -c <scope-name>

For more information, see Secrets.

Step 2: Create a Custom Tecton Environment

Create a custom environment using v1.1 or newer of the tecton-runtime and add the following packages (and other packages you may need) in your RTFV:

1. Create a requirements file: requirements.txt

   tecton-runtime==1.1.0
   pyscopg2-binary
2. Create the environment in Tecton:

   tecton environment create --name my_env --requirements /path/to/requirements.txt

Wait for the environment status to be READY. You can check environment status via the CLI: tecton environment list

You will need the name of this environment when you create your Transform Server Group.

Step 3: Create and Apply a Transform Server Group

Create and apply a Transform Server Group. Be sure to specify the name of the environment you created in Step 2.

A Transform Server Group needs to be applied before a Feature Service can refer to it. Put the following code into a Python file:

from tecton import ProvisionedScalingConfig, TransformServerGroup

postgres_tsg = TransformServerGroup(
    name="my_tsg",
    description="Transform Server Group used for PostgreSQL example",
    environment="my_env",  # name of environment created in Step 2
    scaling_config=ProvisionedScalingConfig(
        desired_nodes=2,
    ),
)

Apply the Transform Server Group:

tecton apply

Wait for the Transform Server Group to be in READY status, check the status with the tecton server-group list CLI command:

% tecton server-group list

               Id                   Name            Type              Status    Environment  ...
================================================================================================
bde2a413a3491a27384fea1234567890   default   TRANSFORM_SERVER_GROUP   READY        None      ...

Step 4: Build the Resource

Provide a name and description for the Resource Provider. Add the scope and key for the Tecton Secret you created in step 1. Your Resource Provider can be applied along with your feature view/etc.

@resource_provider(
    name="pg_rp",
    secrets={"my_pg_token": Secret(scope="<SCOPE_NAME>", key="<KEY_NAME>")},
    description="A postgresql query",
)
def pgclient(context):
    from psycopg2.pool import SimpleConnectionPool

    connection_string = context.secrets["my_pg_token"]
    # Create a connection pool
    connection_pool = SimpleConnectionPool(
        1,  # Minimum number of connections in the pool
        5,  # Maximum number of connections in the pool
        connection_string,
    )
    return connection_pool


request = RequestSource(schema=[Field("id", Int64)])

This resource provider creates a connection pool of 1-5 PostgreSQL database connections that can be shared and reused across multiple queries, which is essential for handling concurrent requests efficiently while preventing resource exhaustion that could occur from creating new connections for every query.

We will apply the Resource Provider in the next step.

Step 5: Finish and Test the System End to End

Step 5a: Create Your Feature View and Resource and Apply Your Code

Here's a partial schema for our PostgreSQL database:

Column Name	Data Type	Constraints
user_id	SERIAL	PRIMARY KEY
first_name	VARCHAR(50)	NOT NULL
last_name	VARCHAR(50)	NOT NULL
phone_number	VARCHAR(15)
balance	NUMERIC(12, 2)	NOT NULL
credit_score	INTEGER

We will return phone_number, balance, and credit_score.

Create the following Python file:

from tecton.types import String, Int64, Field
from tecton import ProvisionedScalingConfig
from tecton import TransformServerGroup, FeatureService
from tecton import resource_provider, Secret, realtime_feature_view, Attribute, RequestSource

# Created and applied in Step 3
#
# postgres_tsg = TransformServerGroup(
#     name="my_tsg",
#     description="Transform Server Group used for PostgreSQL example",
#     environment="my_env",  # name of environment created in Step 2
#     scaling_config=ProvisionedScalingConfig(
#         desired_nodes=2,
#     ),
# )


# Apply the Resource Provider from Step 4
@resource_provider(
    name="pg_rp",
    secrets={"my_pg_token": Secret(scope="<SCOPE_NAME>", key="<KEY_NAME>")},
    description="A postgresql query",
)
def pgclient(context):
    from psycopg2.pool import SimpleConnectionPool

    connection_string = context.secrets["my_pg_token"]
    # Create a connection pool
    connection_pool = SimpleConnectionPool(
        1,  # Minimum number of connections in the pool
        5,  # Maximum number of connections in the pool
        connection_string,
    )
    return connection_pool


request = RequestSource(schema=[Field("id", Int64)])


@realtime_feature_view(
    sources=[request],
    mode="python",
    features=[
        Attribute("phone_number", String),
        Attribute("balance", String),
        Attribute("credit_score", String),
    ],
    resource_providers={"pg_rp": pgclient},
)
def info_fv(request, context):
    client = context.resources["pg_rp"]
    conn = client.getconn()
    # Create a cursor object
    cur = conn.cursor()
    query = "SELECT phone_number, balance, credit_score FROM bank_users WHERE user_id = {id};".format(id=request["id"])
    cur.execute(query)
    user_query_result = cur.fetchone()

    user_data = {}
    if user_query_result:
        user_data["phone_number"] = str(user_query_result[0])
        user_data["balance"] = str(user_query_result[1])
        user_data["credit_score"] = str(user_query_result[2])

    cur.close()
    return user_data


info_service = FeatureService(
    name="user_info_service", online_serving_enabled=True, transform_server_group=my_tsg, features=[info_fv]
)

And apply your code. From the Tecon CLI, run:

tecton apply

Step 5b: Test Your API Resource

Here's a sample curl you can execute to test your feature:

curl -X POST https://<YOUR_CLUSTER>.tecton.ai/api/v1/feature-service/get-features\
     -H "Authorization: Tecton-key <YOUR_SERVICE_ACCOUNT_API_KEY>" -d\
'{
  "params": {
    "feature_service_name": "user_info_service",
    "request_context_map": {
      "id": "67"
    },
    "workspace_name": "<YOUR_WORKSPACE_NAME>",
    "metadataOptions": {
      "includeNames": true
    }
  }
}'

Run the command and you should see output like this:

{"result":{"features":["(628) 604-7767","40513.09","775"]},"metadata":{"features":[{"name":"info_fv.phone_number"},{"name":"info_fv.balance"},{"name":"info_fv.credit_score"}]}}

And that's it -- you have retrieved data directly from a PostgreSQL table into your realtime feature view!

Troubleshooting

If your curl returns an error similar to the following, it could mean there is an issue with your PostgreSQL connection string:

{"error":"OperationalError: SSL connection has been closed unexpectedly\n (when evaluating UDF info_fv)","code":3,"message":"OperationalError: SSL connection has been closed unexpectedly\n (when evaluating UDF info_fv)"}%

If you have verified that your connection string is correct, reach out to Tecton support to configure access to the specified database port.