ydb-dbapi

import os import ydb_dbapi # Configuration for YDB connection # Replace with your YDB endpoint and database path or set as environment variables YDB_ENDPOINT = os.environ.get("YDB_ENDPOINT", "grpc://localhost:2136") YDB_DATABASE = os.environ.get("YDB_DATABASE", "/local") connection = None cursor = None try: # Establish a synchronous connection to YDB connection = ydb_dbapi.connect( host=YDB_ENDPOINT, database=YDB_DATABASE ) print("Successfully connected to YDB.") # Create a cursor object cursor = connection.cursor() # Execute a simple DDL query (create table if not exists) cursor.execute(""" CREATE TABLE IF NOT EXISTS my_table ( id Int64, value Utf8, PRIMARY KEY (id) ); """) print("Table 'my_table' ensured to exist.") connection.commit() # DDL statements usually imply a commit, but explicit is good practice. # Execute an UPSERT statement to insert/update data cursor.execute("UPSERT INTO my_table (id, value) VALUES (1, 'Hello');") cursor.execute("UPSERT INTO my_table (id, value) VALUES (2, 'YDB DBAPI');") connection.commit() # Commit the transaction print("Inserted data into 'my_table'.") # Execute a SELECT statement cursor.execute("SELECT id, value FROM my_table ORDER BY id;") # Fetch all results rows = cursor.fetchall() print("Fetched data:") for row in rows: print(f" ID: {row[0]}, Value: {row[1]}") except ydb_dbapi.Error as e: print(f"An YDB DBAPI error occurred: {e}") if connection: connection.rollback() # Rollback on error except Exception as e: print(f"An unexpected error occurred: {e}") finally: if cursor: cursor.close() if connection: connection.close() print("Connection closed.")