Подключение к базе данных Managed Service for YDB из функции Yandex Cloud Functions на Python

В консоли управления выберите каталог, в котором хотите создать сервисный аккаунт.

В списке сервисов выберите Identity and Access Management.

Нажмите кнопку Создать сервисный аккаунт.

Введите имя сервисного аккаунта, например sa-function. Требования к имени:

• длина — от 3 до 63 символов;
• может содержать строчные буквы латинского алфавита, цифры и дефисы;
• первый символ — буква, последний — не дефис.

Нажмите Добавить роль и выберите editor.

Нажмите кнопку Создать.

В консоли управления выберите каталог, в котором хотите создать БД.

В списке сервисов выберите Managed Service for YDB.

Нажмите кнопку Создать базу данных.

Введите имя БД. Требования к имени:

• длина — от 3 до 63 символов;
• может содержать строчные буквы латинского алфавита, цифры и дефисы;
• первый символ — буква, последний — не дефис.

В блоке Тип базы данных выберите опцию Serverless.

Нажмите кнопку Создать базу данных.

Дождитесь запуска БД. В процессе создания БД будет иметь статус Provisioning. Когда БД станет готова к использованию, статус сменится на Running.

Нажмите на имя созданной БД.

В блоке Соединение найдите поле Эндпоинт и сохраните его значение. Оно понадобится на следующем шаге.

В консоли управления выберите каталог, в котором хотите создать функцию.

В списке сервисов выберите Cloud Functions.

Нажмите кнопку Создать функцию.

Введите имя и описание функции. Требования к имени:

• длина — от 3 до 63 символов;
• может содержать строчные буквы латинского алфавита, цифры и дефисы;
• первый символ — буква, последний — не дефис.

Нажмите кнопку Создать.

В блоке Редактор выберите среду выполнения Python, отключите опцию Добавить файлы с примерами кода и нажмите Продолжить.

В блоке Код функции создайте файл index.py и вставьте в него следующий код:

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

В блоке Код функции создайте файл requirements.txt и вставьте в него следующий текст:

ydb

В качестве точки входа укажите index.handler.

Выберите сервисный аккаунт, например sa-function.

Настройте переменные окружения:

• YDB_ENDPOINT — введите первую часть сохраненного ранее поля Эндпоинт (часть до вхождения /?database=). Например, grpcs://ydb.serverless.yandexcloud.net:2135.
• YDB_DATABASE — введите вторую часть сохраненного ранее поля Эндпоинт (часть после вхождения /?database=). Например, /ru-central1/r1gra875baom********/g5n22e7ejfr1********.

Нажмите кнопку Сохранить изменения.

Перейдите на вкладку Тестирование.

Нажмите кнопку Запустить тест и посмотрите результат тестирования.

При успешном подключении к БД и выполнении запроса состояние функции изменится на Выполнена, а ответ функции будет содержать следующий текст:

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