Connecting to a Managed Service for YDB database from a Python function in Yandex Cloud Functions

In the management console, select the folder where you want to create your service account.

In the list of services, select Identity and Access Management.

Click Create service account.

Enter a name for the service account, e.g., sa-function. The naming requirements are as follows:

• It must be from 2 to 63 characters long.
• It may contain lowercase Latin letters, numbers, and hyphens.
• It must start with a letter and cannot end with a hyphen.

Click Add role and select editor.

Click Create.

In the management console, select the folder where you want to create a database.

In the list of services, select Managed Service for YDB.

Click Create a database.

Name the database. The naming requirements are as follows:

• It must be from 2 to 63 characters long.
• It may contain lowercase Latin letters, numbers, and hyphens.
• It must start with a letter and cannot end with a hyphen.

Under Database type, select Serverless.

Click Create a database.

Wait for the DB to start. While being created, the database will have the Provisioning status. Once it is ready for use, its status will change to Running.

Click the name of the database you created.

Under Connection, find the Endpoint field and save its value. You will need it at the next step.

In the Management console, select the folder you want to create a function in.

In the list of services, select Cloud Functions.

Click Create function.

Enter a name and description for the function. The naming requirements are as follows:

• It must be from 2 to 63 characters long.
• It may contain lowercase Latin letters, numbers, and hyphens.
• It must start with a letter and cannot end with a hyphen.

Click Create.

Under Editor, select Python as the runtime environment, disable Add files with code examples, and click Continue.

Under Function code, create a file named index.py and paste the following code into it:

import os
import ydb
import ydb.iam

# Create driver in global space.
driver = ydb.Driver(
  endpoint=os.getenv('YDB_ENDPOINT'),
  database=os.getenv('YDB_DATABASE'),
  credentials=ydb.iam.MetadataUrlCredentials(),
)

# Wait for the driver to become active for requests.

driver.wait(fail_fast=True, timeout=5)

# Create the session pool instance to manage YDB sessions.
pool = ydb.SessionPool(driver)

def execute_query(session):
  # Create the transaction and execute query.
  return session.transaction().execute(
    'select 1 as cnt;',
    commit_tx=True,
    settings=ydb.BaseRequestSettings().with_timeout(3).with_operation_timeout(2)
  )

def handler(event, context):
  # Execute query with the retry_operation helper.
  result = pool.retry_operation_sync(execute_query)
  return {
    'statusCode': 200,
    'body': str(result[0].rows[0].cnt == 1),
  }

Under Function code, create a file named requirements.txt and paste the following text into it:

ydb

Specify the entry point: index.handler.

Select a service account, e.g., sa-function.

Configure the environment variables:

• YDB_ENDPOINT: Enter the first part of the previously saved Endpoint field value (preceding /?database=), e.g., grpcs://ydb.serverless.yandexcloud.net:2135.
• YDB_DATABASE: Enter the second part of the previously saved Endpoint field value (following /?database=), e.g., /ru-central1/r1gra875baom********/g5n22e7ejfr1********.

Click Save changes.

Navigate to the Testing tab.

Click Run test and check out the testing results.

If a DB connection is established and a query is executed, the function status will change to Done and its output will contain the following text:

{
  "statusCode": 200,
  "body": "True"
}