SQLAlchemy 允许在声明
String
列时指定长度:
foo = Column(String(10))
如 SQL 所示:
foo VARCHAR(10)
我知道某些 DBMS 在表中创建行时使用此长度值来分配内存。但有些 DBMS(如 SQLite)不关心它,仅为了与 SQL 标准兼容而接受此语法。但有些 DBMS(如 MySQL)要求指定它。
就我个人而言,我喜欢指定某些文本数据的最大长度,因为它有助于设计 UI,因为您知道显示它所需的区域。
此外,我认为这将使我的应用程序行为在不同的 DBMS 中更加一致。
因此,我想通过根据声明的长度检查其长度(当声明长度时)来验证插入时字符串/Unicode 列的值。
第一个解决方案是使用检查约束:
from sqlalchemy import CheckConstraint, Column, Integer, String, create_engine
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
engine = create_engine("sqlite:///:memory:", echo=True)
Base = declarative_base(bind=engine)
Session = sessionmaker(bind=engine)
class Foo(Base):
__tablename__ = "Foo"
id = Column(Integer, primary_key=True)
bar = Column(String(10), CheckConstraint("LENGTH(bar) < 10"))
Base.metadata.create_all()
if __name__ == "__main__":
session = Session()
session.add(Foo(bar="a" * 20))
try:
session.commit()
except IntegrityError as e:
print(f"Failed with: {e.orig}")
它可以工作,但 SQL 约束表达式不是由 SQLAlchemy 生成的。因此,如果 DBMS 需要不同的语法,则可能需要一些自定义生成。
我还尝试使用 SQLAlchemy validator:
class Foo(Base):
__tablename__ = "Foo"
id = Column(Integer, primary_key=True)
bar = Column(String(10))
@validates("bar")
def check_bar_length(self, key, value):
column_type = getattr(type(self), key).expression.type
max_length = column_type.length
if len(value) > max_length:
raise ValueError(
f"Value '{value}' for column '{key}' "
f"exceed maximum length of '{max_length}'"
)
return value
try:
Foo(bar="a" * 20)
except ValueError as e:
print(f"Failed with: {e}")
现在,最大长度是根据声明的长度推断出来的。
检查是在实体创建时完成的,而不是在提交时完成的。不知道会不会有问题。
上面显示的两种解决方案都需要对每列应用验证。我正在寻找一种解决方案来自动检查具有声明长度的 String/Unicode 列。
使用自定义类型可能是解决方案。但这看起来像是一个丑陋的黑客,因为自定义类型不是为了数据验证而是为了数据转换。
那么,您是否考虑另一种解决方案,也许是我不知道的 SQLAlchemy 功能,这将帮助我将检查自动添加到指定了
String
的所有 length
列?
另一个选项可能是显式定义表并分解字符串列定义,以便为每个字符串列创建检查约束,而无需重复它。
def string_column(name, length):
check_str = "LENGTH({}) < {}".format(name, length)
return Column(name, String(length), CheckConstraint(check_str))
class Foo(Base):
__table__ = Table("Foo", Base.metadata,
Column("id", Integer, primary_key=True),
string_column("bar", 10),
string_column("name", 15))
我找到了一个似乎适合我的需求的解决方案。但我认为我添加约束的方式有点hacky。
它涉及到以下用途:
实体像往常一样声明,不需要指定任何约束:
from sqlalchemy import Column, Integer, LargeBinary, String, Unicode,
class Foo(Entity):
__tablename__ = "Foo"
id = Column(Integer, primary_key=True)
string_without_length = Column(String())
string_with_length = Column(String(10))
unicode_with_length = Column(Unicode(20))
binary = Column(LargeBinary(256))
在对类进行检测之前将约束附加到列:
from sqlalchemy import CheckConstraint, func, String
from sqlalchemy.event import listen_for
from sqlalchemy.orm import mapper
@listens_for(mapper, "instrument_class")
def add_string_length_constraint(mapper, cls):
table = cls.__table__
for column in table.columns:
if isinstance(column.type, String):
length = column.type.length
if length is not None:
CheckConstraint(
func.length(column) <= length,
table=column,
_autoattach=False,
)
CREATE TABLE "Foo" (
id INTEGER NOT NULL,
string_without_length VARCHAR,
string_with_length VARCHAR(10) CHECK (length(string_with_length) <= 10),
unicode_with_length VARCHAR(20) CHECK (length(unicode_with_length) <= 20),
binary BLOB,
PRIMARY KEY (id)
)
String
没有长度的列不受影响,String
和 Unicode
列添加了 CHECK 约束,length
参数的其他列(如 LargeBinary)不受影响。@listens_for(mapper, "instrument_class")
当创建了检测类的映射器但未完全初始化时,会发生
instrument_class
事件。它可以在您的基本声明类(使用 declarative_base()
创建)上或直接在 slqalchemy.orm.mapper
类上收听。
if isinstance(column.type, String):
仅
String
(以及像Unicode
这样的子类)列...
if length is not None:
...设置了
length
的人会被考虑。
CheckConstraint(
func.length(column) <= length,
table=column,
_autoattach=False,
)
约束是使用 SQLAlchemy 表达式生成的。
最后是hacky部分:
创建约束时,SQLAlchemy 自动将其附加到表(我认为它检测到约束所涉及的列)。
由于我希望将其生成为列定义的一部分,因此我使用
_autoattach=False
禁用此自动附加,然后使用 table=column
指定列。
如果你不关心它,请忽略这些论点:
CheckConstraint(func.length(column) <= length)
生成的 DDL 语句将是:
CREATE TABLE "Foo" (
id INTEGER NOT NULL,
string_without_length VARCHAR,
string_with_length VARCHAR(10),
unicode_with_length VARCHAR(20),
binary BLOB,
PRIMARY KEY (id),
CHECK (length(string_with_length) <= 10),
CHECK (length(unicode_with_length) <= 20)
)
我不喜欢由数据库引擎完成检查,因为当它发生时,损坏就会完成,并且我会收到操作错误。我认为最好在插入之前进行检查,以某种方式减轻损坏并打印日志消息,以便我可以在应用程序中发生实际错误时修复它。下面是我的想法(我最近从一个默默地截断字符串的数据库迁移到一个在字符串太长时抛出错误的数据库):
from sqlalchemy import Column, Integer, CHAR, String, event
from sqlalchemy.orm import relationship, declarative_base
Base = declarative_base()
class Foo(Base):
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
s1 = Column(String(5))
s2 = Column(CHAR(5))
def add_string_truncaters(base):
def add_listener(column, length):
@event.listens_for(column, 'set', retval=True)
def truncate(target, value, oldvalue, initiator):
if len(value) > length:
print(f'Truncated {column}')
return value[:length]
for cls in globals().values():
if (not isinstance(cls, type)
or cls is base
or not issubclass(cls, base)):
continue
for colname in cls.__table__.columns.keys():
col = getattr(cls, colname, None)
if col and isinstance(col.type, String):
add_listener(col, col.type.length)
add_string_truncaters(Base)
x = Foo(s1='abcdefgh', s2='abcdefgh')
print(x.s1, x.s2)
看起来不漂亮,但听起来不错。