Open
Description
Hello!
I encountered an issue where I can create tables with native UUID fields, but I cannot insert records into tables with such fields.
I noticed that native UUID support is not officially provided yet, but I decided to dig deeper and found a temporary workaround that I want to share with you.
UUIDs can be inserted if, during SQL statement construction, they are processed as
CAST(<uuid> AS UUID) ?? Uuid('00000000-0000-0000-0000-000000000000')
Accordingly, I wrote a solution that allows using native UUID in YDB instead of UTF8 strings as UUIDs.
Dependencies:
pyproject.toml
[project]
name = "ydb-test"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"advanced-alchemy>=1.4.4",
"sqlalchemy>=2.0.41",
"uuid-utils>=0.11.0",
"ydb-sqlalchemy>=0.1.8",
]To install, use the command uv lock && uv sync
Here is a minimal example to help you quickly get into the problem
import asyncio
import random
import uuid
from typing import Any
from sqlalchemy import String, AsyncAdaptedQueuePool, insert, select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import UUID
from uuid_utils import uuid7
HOST='localhost'
PORT=2136
NAME='local'
# База и движок (замените на вашу строку подключения к YDB)
connection_args: dict[str, Any] = {
"paramstyle": "named",
}
engine = create_async_engine(
f"yql+ydb_async://{HOST}:{PORT}/{NAME}",
echo=True,
poolclass=AsyncAdaptedQueuePool,
_add_declare_for_yql_stmt_vars=True,
connect_args=connection_args,
)
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase):
pass
class TestTable(Base):
__tablename__ = "test_table"
id: Mapped[uuid.UUID] = mapped_column(UUID, primary_key=True, default=uuid7)
title: Mapped[str] = mapped_column(String(100))
some_other_id: Mapped[uuid.UUID] = mapped_column(UUID, primary_key=True, default=uuid7)
async def main() -> None:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
async with async_session_maker() as session:
for _ in range(20):
stmt = insert(TestTable).values(title="Test title" + str(random.randint(1,999)))
await session.execute(stmt)
await session.commit()
fetched = (await session.execute(select(TestTable))).scalars()
for row in fetched:
print(f"Fetched record: id={row.id}, title={row.title}")
if __name__ == '__main__':
asyncio.run(main())And here is my temporary solution that enabled using native UUIDs instead of string representations
import asyncio
import random
import uuid
from typing import Any,Optional
from sqlalchemy import String, AsyncAdaptedQueuePool, insert, TypeDecorator, Dialect, select, text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import UUID
from sqlalchemy.sql.type_api import _BindProcessorType
from sqlalchemy.sql.sqltypes import _UUID_RETURN
from uuid_utils import uuid7
HOST='localhost'
PORT=2136
NAME='local'
connection_args: dict[str, Any] = {
"paramstyle": "named",
}
engine = create_async_engine(
f"yql+ydb_async://{HOST}:{PORT}/{NAME}",
echo=True,
poolclass=AsyncAdaptedQueuePool,
_add_declare_for_yql_stmt_vars=True,
connect_args=connection_args,
)
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
class YqlUUIDImpl(UUID):
def bind_processor(
self, dialect: Dialect
) -> Optional[_BindProcessorType[_UUID_RETURN]]:
def process(value):
return str(value)
return process
def result_processor(self, dialect, coltype):
def process(value):
if value is None:
return value
if isinstance(value, uuid.UUID):
return value
return value
return process
class YdbUUID(TypeDecorator):
impl = YqlUUIDImpl
def process_bind_param(self, value, dialect):
if value is None:
return None
if isinstance(value, uuid.UUID):
return str(value) # строка с дефисами
return value # UUID строка с дефисами
def process_result_value(self, value, dialect):
if value is None:
return None
if isinstance(value, uuid.UUID):
return str(value)
return uuid.UUID(value)
def bind_expression(self, bindvalue):
param_name = bindvalue._orig_key
sql_text = f"CAST(:{param_name} AS UUID) ?? Uuid('00000000-0000-0000-0000-000000000000')"
return text(sql_text).bindparams(bindvalue)
class Base(DeclarativeBase):
pass
class TestTable(Base):
__tablename__ = "test_table"
id: Mapped[uuid.UUID] = mapped_column(YdbUUID, primary_key=True, default=uuid7)
title: Mapped[str] = mapped_column(String(100))
some_other_id: Mapped[uuid.UUID] = mapped_column(UUID, primary_key=True, default=uuid7)
async def main() -> None:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
async with async_session_maker() as session:
for _ in range(20):
stmt = insert(TestTable).values(title="Test title" + str(random.randint(1,999)))
await session.execute(stmt)
await session.commit()
fetched = (await session.execute(select(TestTable))).scalars()
for row in fetched:
print(f"Fetched record: id={row.id}, title={row.title}")
if __name__ == '__main__':
asyncio.run(main())Please, if you have the opportunity, kindly add at least support for a type that casts a string to UUID instead of UUID as UTF8. That would really make life much easier! 🙌
If you want me to help format the example or anything else, just let me know!
Metadata
Metadata
Assignees
Labels
No labels