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.

Go to Identity and Access Management.

Click Create service account.

Enter a name for the service account, e.g., sa-function. Follow these naming requirements:

• Length: between 3 and 63 characters.
• It can only 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.

Go to Managed Service for YDB.

Click Create a database.

Name the database. Follow these naming requirements:

• Length: between 3 and 63 characters.
• It can only 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 database 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 database name.

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 where you want to create a function.

Go to Cloud Functions.

Click Create function.

Enter a name and description for the function. Follow these naming requirements:

• Length: between 3 and 63 characters.
• It can only 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 test results.

If your database is connected and successfully queried, the function status will change to Done and its output will contain the following text:

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