如何在SQLAlchemy中使用UUID



如果使用PostgreSQL(Postgres),是否有方法在SQLAlchemy中将列(主键)定义为UUID?

sqlalchemy-postgres方言支持UUID列。这很简单(这个问题特别是postgres)——我不明白为什么其他答案都这么复杂。

这里有一个例子:

from sqlalchemy.dialects.postgresql import UUID
from flask_sqlalchemy import SQLAlchemy
import uuid
db = SQLAlchemy()
class Foo(db.Model):
    id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)

注意不要错过将callable uuid.uuid4传递到列定义中,而不是使用uuid.uuid4()调用函数本身。否则,该类的所有实例都将具有相同的标量值。更多详细信息请点击此处:

标量、Python可调用或ColumnElement表达式,表示此列的默认值,如果插入的VALUES子句中未指定此列,则插入时将调用该表达式。

我写了这个,域不见了,但这里有勇气。。。。

不管我那些真正关心正确数据库设计的同事对用于关键字段的UUID和GUID有何看法。我经常发现我需要这样做。我认为它比自动增量有一些优势,这使它值得。

在过去的几个月里,我一直在完善UUID列类型,我想我终于把它弄得很好了。

from sqlalchemy import types
from sqlalchemy.dialects.mysql.base import MSBinary
from sqlalchemy.schema import Column
import uuid

class UUID(types.TypeDecorator):
    impl = MSBinary
    def __init__(self):
        self.impl.length = 16
        types.TypeDecorator.__init__(self,length=self.impl.length)
    def process_bind_param(self,value,dialect=None):
        if value and isinstance(value,uuid.UUID):
            return value.bytes
        elif value and not isinstance(value,uuid.UUID):
            raise ValueError,'value %s is not a valid uuid.UUID' % value
        else:
            return None
    def process_result_value(self,value,dialect=None):
        if value:
            return uuid.UUID(bytes=value)
        else:
            return None
    def is_mutable(self):
        return False

id_column_name = "id"
def id_column():
    import uuid
    return Column(id_column_name,UUID(),primary_key=True,default=uuid.uuid4)
# Usage
my_table = Table('test',
         metadata,
         id_column(),
         Column('parent_id',
            UUID(),
            ForeignKey(table_parent.c.id)))

我认为二进制(16字节)存储最终应该比字符串表示(36字节?)更高效,而且似乎有迹象表明,在mysql中索引16字节块应该比字符串更高效。我没想到会更糟。

我发现的一个缺点是,至少在phpymyadmin中,您不能编辑记录,因为它隐式地试图对"select*fromtablewhere id=…"进行某种字符转换,并且存在其他显示问题。

除此之外,一切似乎都很好,所以我把它扔掉了。如果你看到一个明显的错误,请留下评论。我欢迎任何改进建议。

除非我遗漏了什么,否则如果底层数据库具有UUID类型,上述解决方案将起作用。如果没有,那么在创建表时可能会出现错误。我提出的解决方案最初是针对MSSqlServer的,最后是MySql,所以我认为我的解决方案更灵活一些,因为它似乎在MySql和sqlite上运行良好。还没检查postgres。

如果您对具有UUID值的"String"列感到满意,这里有一个简单的解决方案:

def generate_uuid():
    return str(uuid.uuid4())
class MyTable(Base):
    __tablename__ = 'my_table'
    uuid = Column(String, name="uuid", primary_key=True, default=generate_uuid)

我使用了SQLAlchemy-Utils包中的UUIDType

由于您使用的是Postgres,这应该可以工作:

from app.main import db
from sqlalchemy.dialects.postgresql import UUID
class Foo(db.Model):
    id = db.Column(UUID(as_uuid=True), primary_key=True)
    name = db.Column(db.String, nullable=False)

这是一种基于SQLAlchemy文档中后端不可知GUID的方法,但使用BINARY字段将UUID存储在非postgresql数据库中。

import uuid
from sqlalchemy.types import TypeDecorator, BINARY
from sqlalchemy.dialects.postgresql import UUID as psqlUUID
class UUID(TypeDecorator):
    """Platform-independent GUID type.
    Uses Postgresql's UUID type, otherwise uses
    BINARY(16), to store UUID.
    """
    impl = BINARY
    def load_dialect_impl(self, dialect):
        if dialect.name == 'postgresql':
            return dialect.type_descriptor(psqlUUID())
        else:
            return dialect.type_descriptor(BINARY(16))
    def process_bind_param(self, value, dialect):
        if value is None:
            return value
        else:
            if not isinstance(value, uuid.UUID):
                if isinstance(value, bytes):
                    value = uuid.UUID(bytes=value)
                elif isinstance(value, int):
                    value = uuid.UUID(int=value)
                elif isinstance(value, str):
                    value = uuid.UUID(value)
        if dialect.name == 'postgresql':
            return str(value)
        else:
            return value.bytes
    def process_result_value(self, value, dialect):
        if value is None:
            return value
        if dialect.name == 'postgresql':
            return uuid.UUID(value)
        else:
            return uuid.UUID(bytes=value)

如果有人感兴趣,我一直在使用Tom Willis的答案,但发现向uuid添加字符串很有用。process_bind_param方法中的UUID转换

class UUID(types.TypeDecorator):
    impl = types.LargeBinary
    def __init__(self):
        self.impl.length = 16
        types.TypeDecorator.__init__(self, length=self.impl.length)
    def process_bind_param(self, value, dialect=None):
        if value and isinstance(value, uuid.UUID):
            return value.bytes
        elif value and isinstance(value, basestring):
            return uuid.UUID(value).bytes
        elif value:
            raise ValueError('value %s is not a valid uuid.UUId' % value)
        else:
            return None
    def process_result_value(self, value, dialect=None):
        if value:
            return uuid.UUID(bytes=value)
        else:
            return None
    def is_mutable(self):
        return False

我遇到了同样的问题,这应该有效,它对我有效:

from sqlalchemy import Column, text
from sqlalchemy.dialects.postgresql import UUID
Column(
    "id", UUID(as_uuid=True),
    primary_key=True,
    server_default=text("gen_random_uuid()"),
)

如果您使用PostgreSQL<14,我想你需要添加这个扩展包:

CREATE EXTENSION IF NOT EXISTS "pgcrypto";

你也可以使用uuid_generate_v4(),你需要添加EXTENSION包,然后:

CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

SQLAlchemy 2.0添加了UUID类型,这是数据库不可知类型的SQL原生形式,与以前的仅PostgreSQL版本的UUID向后兼容。

示例:

import sqlalchemy as sa
from sqlalchemy.orm import DeclarativeBase, Mapped
class Base(DeclarativeBase):
    pass
class MyModel(Base):
    my_field: Mapped[sa.UUID]

我们可以使用UUIDType

from sqlalchemy_utils import UUIDType
from sqlalchemy import String
class User(Base):
    id = Column(UUIDType(binary=False), primary_key=True, default=uuid.uuid4)
    name = Column(String)

有关更多详细信息,我们可以参考官方文件。

您可以尝试编写自定义类型,例如:

import sqlalchemy.types as types
class UUID(types.TypeEngine):
    def get_col_spec(self):
        return "uuid"
    def bind_processor(self, dialect):
        def process(value):
            return value
        return process
    def result_processor(self, dialect):
        def process(value):
            return value
        return process
table = Table('foo', meta,
    Column('id', UUID(), primary_key=True),
)

相关内容

  • 没有找到相关文章

最新更新