Officially supported Python client for YDB distributed SQL database
Complete type system mapping Python types to YDB types, including primitives, optionals, containers, and advanced types like decimals and variants.
Basic data types that map directly to YDB's primitive types.
class PrimitiveType:
"""YDB primitive data types."""
# Boolean
Bool: Type
# Integer types
Int8: Type # Signed 8-bit integer
Uint8: Type # Unsigned 8-bit integer
Int16: Type # Signed 16-bit integer
Uint16: Type # Unsigned 16-bit integer
Int32: Type # Signed 32-bit integer
Uint32: Type # Unsigned 32-bit integer
Int64: Type # Signed 64-bit integer
Uint64: Type # Unsigned 64-bit integer
# Floating point
Float: Type # 32-bit floating point
Double: Type # 64-bit floating point
# String types
String: Type # Binary string
Utf8: Type # UTF-8 string
# Structured data
Yson: Type # YSON format
Json: Type # JSON string
JsonDocument: Type # JSON document (binary)
# UUID
Uuid: Type # UUID type
# Date and time
Date: Type # Date only
Datetime: Type # Date and time
Timestamp: Type # Timestamp with microseconds
Interval: Type # Time interval
TzDate: Type # Date with timezone
TzDatetime: Type # DateTime with timezone
TzTimestamp: Type # Timestamp with timezone
# Decimal number
DyNumber: Type # Dynamic numberNullable types that can contain either a value or NULL.
def Optional(item_type: Type) -> OptionalType:
"""
Create optional (nullable) type.
Args:
item_type (Type): Inner type that can be null
Returns:
OptionalType: Optional type wrapper
"""
class OptionalType:
def __init__(self, item_type: Type):
"""
Optional type constructor.
Args:
item_type (Type): Type that can be null
"""
@property
def item_type(self) -> Type:
"""Get the inner type."""Collection types for complex data structures.
def List(item_type: Type) -> ListType:
"""
Create list type.
Args:
item_type (Type): Type of list elements
Returns:
ListType: List type
"""
def Tuple(*item_types: Type) -> TupleType:
"""
Create tuple type with fixed element types.
Args:
*item_types (Type): Types of tuple elements in order
Returns:
TupleType: Tuple type
"""
def Struct(**kwargs: Type) -> StructType:
"""
Create struct type with named fields.
Args:
**kwargs (Type): Field names and their types
Returns:
StructType: Struct type
"""
def Dict(key_type: Type, value_type: Type) -> DictType:
"""
Create dictionary type.
Args:
key_type (Type): Type of dictionary keys
value_type (Type): Type of dictionary values
Returns:
DictType: Dictionary type
"""
class ListType:
def __init__(self, item_type: Type):
"""List type with homogeneous elements."""
@property
def item_type(self) -> Type:
"""Get element type."""
class TupleType:
def __init__(self, *item_types: Type):
"""Tuple type with heterogeneous elements."""
@property
def item_types(self) -> tuple:
"""Get element types."""
class StructType:
def __init__(self, **kwargs: Type):
"""Struct type with named fields."""
@property
def fields(self) -> dict:
"""Get field definitions."""
class DictType:
def __init__(self, key_type: Type, value_type: Type):
"""Dictionary type with key-value pairs."""
@property
def key_type(self) -> Type:
"""Get key type."""
@property
def value_type(self) -> Type:
"""Get value type."""Specialized types for complex scenarios.
def Decimal(precision: int, scale: int) -> DecimalType:
"""
Create decimal number type.
Args:
precision (int): Total number of digits
scale (int): Number of digits after decimal point
Returns:
DecimalType: Decimal type
"""
def Variant(struct_type: StructType = None, tuple_type: TupleType = None) -> VariantType:
"""
Create variant (union) type.
Args:
struct_type (StructType, optional): Struct-based variant
tuple_type (TupleType, optional): Tuple-based variant
Returns:
VariantType: Variant type
"""
def Tagged(tag: str, base_type: Type) -> TaggedType:
"""
Create tagged type with metadata.
Args:
tag (str): Type tag/label
base_type (Type): Underlying type
Returns:
TaggedType: Tagged type
"""
class DecimalType:
def __init__(self, precision: int, scale: int):
"""
Decimal number type.
Args:
precision (int): Total digits (1-35)
scale (int): Decimal places (0-precision)
"""
@property
def precision(self) -> int:
"""Total number of digits."""
@property
def scale(self) -> int:
"""Digits after decimal point."""
class VariantType:
def __init__(self, underlying_type: Type):
"""Variant type for union values."""
@property
def underlying_type(self) -> Type:
"""Get underlying type."""
class TaggedType:
def __init__(self, tag: str, base_type: Type):
"""Tagged type with metadata."""
@property
def tag(self) -> str:
"""Get type tag."""
@property
def base_type(self) -> Type:
"""Get base type."""
class VoidType:
"""Void type for functions with no return value."""Helper functions for working with types.
class Type:
"""Base class for all YDB types."""
def __str__(self) -> str:
"""String representation of type."""
def __eq__(self, other) -> bool:
"""Type equality comparison."""
def type_to_native(ydb_type: Type):
"""
Convert YDB type to native Python representation.
Args:
ydb_type (Type): YDB type instance
Returns:
Native Python representation
"""
def native_to_type(python_value):
"""
Infer YDB type from Python value.
Args:
python_value: Python value
Returns:
Type: Corresponding YDB type
"""import ydb
# Primitive types
user_id_type = ydb.PrimitiveType.Uint64
name_type = ydb.PrimitiveType.Utf8
created_at_type = ydb.PrimitiveType.Timestamp
is_active_type = ydb.PrimitiveType.Bool
# Optional types (nullable)
optional_email = ydb.Optional(ydb.PrimitiveType.Utf8)
optional_age = ydb.Optional(ydb.PrimitiveType.Uint32)import ydb
# List of strings
tags_type = ydb.List(ydb.PrimitiveType.Utf8)
# Dictionary mapping strings to integers
scores_type = ydb.Dict(ydb.PrimitiveType.Utf8, ydb.PrimitiveType.Int32)
# Tuple with mixed types
coordinate_type = ydb.Tuple(ydb.PrimitiveType.Double, ydb.PrimitiveType.Double)
# Struct with named fields
user_type = ydb.Struct(
id=ydb.PrimitiveType.Uint64,
name=ydb.PrimitiveType.Utf8,
email=ydb.Optional(ydb.PrimitiveType.Utf8),
tags=ydb.List(ydb.PrimitiveType.Utf8)
)import ydb
# Decimal for precise monetary calculations
price_type = ydb.Decimal(precision=10, scale=2) # 10 digits, 2 decimal places
# Variant type for union of different types
status_type = ydb.Variant(ydb.Struct(
success=ydb.PrimitiveType.Bool,
error=ydb.PrimitiveType.Utf8,
pending=ydb.PrimitiveType.Void
))
# Tagged type for semantic meaning
user_id_tagged = ydb.Tagged("UserId", ydb.PrimitiveType.Uint64)import ydb
# Define table with various column types
table_desc = ydb.TableDescription() \
.with_column(ydb.TableColumn(
'id',
ydb.PrimitiveType.Uint64
)) \
.with_column(ydb.TableColumn(
'profile',
ydb.Struct(
name=ydb.PrimitiveType.Utf8,
age=ydb.Optional(ydb.PrimitiveType.Uint32),
emails=ydb.List(ydb.PrimitiveType.Utf8)
)
)) \
.with_column(ydb.TableColumn(
'metadata',
ydb.Dict(ydb.PrimitiveType.Utf8, ydb.PrimitiveType.Utf8)
)) \
.with_primary_key('id')import ydb
from decimal import Decimal
from datetime import datetime, date
# Python to YDB type inference
python_values = {
42: ydb.PrimitiveType.Int32,
"hello": ydb.PrimitiveType.Utf8,
True: ydb.PrimitiveType.Bool,
3.14: ydb.PrimitiveType.Double,
datetime.now(): ydb.PrimitiveType.Timestamp,
date.today(): ydb.PrimitiveType.Date,
[1, 2, 3]: ydb.List(ydb.PrimitiveType.Int32),
{"key": "value"}: ydb.Dict(ydb.PrimitiveType.Utf8, ydb.PrimitiveType.Utf8)
}
# Working with optional values
def process_optional_field(value):
if value is None:
return None # NULL value
return value # Actual valueimport ydb
def typed_query_example(session):
# Use typed parameters in queries
result_sets = session.transaction().execute(
"""
DECLARE $user_id AS Uint64;
DECLARE $profile AS Struct<
name: Utf8,
age: Uint32?,
emails: List<Utf8>
>;
INSERT INTO users (id, profile)
VALUES ($user_id, $profile);
""",
parameters={
'$user_id': 123,
'$profile': {
'name': 'Alice',
'age': 30,
'emails': ['alice@example.com', 'alice.work@company.com']
}
},
commit_tx=True
)import ydb
# Define complex nested structure
order_type = ydb.Struct(
order_id=ydb.PrimitiveType.Uint64,
customer=ydb.Struct(
id=ydb.PrimitiveType.Uint64,
name=ydb.PrimitiveType.Utf8,
email=ydb.Optional(ydb.PrimitiveType.Utf8)
),
items=ydb.List(ydb.Struct(
product_id=ydb.PrimitiveType.Uint64,
quantity=ydb.PrimitiveType.Uint32,
price=ydb.Decimal(precision=10, scale=2)
)),
status=ydb.Variant(ydb.Struct(
pending=ydb.PrimitiveType.Void,
shipped=ydb.PrimitiveType.Timestamp,
delivered=ydb.PrimitiveType.Timestamp,
cancelled=ydb.PrimitiveType.Utf8 # cancellation reason
)),
metadata=ydb.Optional(ydb.Dict(
ydb.PrimitiveType.Utf8,
ydb.PrimitiveType.Utf8
))
)
# Use in table definition
orders_table = ydb.TableDescription() \
.with_column(ydb.TableColumn('order', order_type)) \
.with_primary_key('order.order_id')from typing import Union, Any, Dict, List as PyList
from decimal import Decimal
from datetime import datetime, date, timedelta
# Type mapping helpers
PythonValue = Union[None, bool, int, float, str, bytes, datetime, date, timedelta, Decimal, PyList, Dict[str, Any]]
YdbValue = Any
# Time units for intervals
class Unit:
SECONDS: Unit
MILLISECONDS: Unit
MICROSECONDS: Unit
NANOSECONDS: Unit