将 SQLAlchemy hybrid_property 与本机 属性 构造相结合
Combine SQLAlchemy hybrid_property with native property construction
我在 SQLAlchemy 中有一个用户 class。我希望能够在数据库中加密用户的电子邮件地址属性，但仍然可以通过过滤器查询对其进行搜索。
我的问题是，如果我使用@hybrid_property，我的查询理论上可以工作，但我的构造没有，如果我使用@属性，我的构造可以工作，但我的查询没有
from cryptography.fernet import Fernet # <- pip install cryptography
from werkzeug.security import generate_password_hash
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email_hash = db.Column(db.String(184), unique=True, nullable=False)
password_hash = db.Column(db.String(128))
# @property # <- Consider this as option 2...
@hybrid_property # <- Consider this as option 1...
def email(self):
f = Fernet('SOME_ENC_KEY')
value = f.decrypt(self.email_hash.encode('utf-8'))
return value
@email.setter
def email(self, email):
f = Fernet('SOME_ENC_KEY')
self.email_hash = f.encrypt(email.encode('utf-8'))
@property
def password(self):
raise AttributeError('password is not a readable attribute.')
@password.setter
def password(self, password):
self.password_hash = generate_password_hash(password)
def __init__(self, **kwargs):
super(User, self).__init__(**kwargs)
# other checks and modifiers
对于选项 1：当我尝试使用 User(email='a@example.com',password='secret') 构建用户时，我收到回溯，
~/models.py in __init__(self, **kwargs)
431 # Established role assignment by default class initiation
432 def __init__(self, **kwargs):
--> 433 super(User, self).__init__(**kwargs)
434 if self.role is None:
435 _default_role = Role.query.filter_by(default=True).first()
~/lib/python3.6/site-packages/sqlalchemy/ext/declarative/base.py in _declarative_constructor(self, **kwargs)
697 raise TypeError(
698 "%r is an invalid keyword argument for %s" %
--> 699 (k, cls_.__name__))
700 setattr(self, k, kwargs[k])
701 _declarative_constructor.__name__ = '__init__'
TypeError: 'email' is an invalid keyword argument for User
对于选项 2：如果我改为将 @hybrid_property 更改为 @属性 构造没问题，但是我的查询 User.query.filter_by(email=form.email.data.lower()).first() 失败并且 returns None.
我应该更改什么才能使其按要求工作？
==============
请注意，我应该说我已经尽量避免使用双重属性，因为我不想对底层代码库进行大量编辑。所以我明确地试图避免根据 User(email_input='a@a.com', password='secret') 和 User.query.filter_by(email='a@a.com').first():
将创建与查询分开
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email_hash = db.Column(db.String(184), unique=True, nullable=False)
password_hash = db.Column(db.String(128))
@hybrid_property
def email(self):
f = Fernet('SOME_ENC_KEY')
value = f.decrypt(self.email_hash.encode('utf-8'))
return value
@property
def email_input(self):
raise AttributeError('email_input is not a readable attribute.')
@email_input.setter
def email_input(self, email):
f = Fernet('SOME_ENC_KEY')
self.email_hash = f.encrypt(email.encode('utf-8'))
@property
def password(self):
raise AttributeError('password is not a readable attribute.')
@password.setter
def password(self, password):
self.password_hash = generate_password_hash(password)
def __init__(self, **kwargs):
super(User, self).__init__(**kwargs)
# other checks and modifiers
在您的 hybrid_property、email 中，如果 self.email_hash 是 str 类型，则行 self.f.decrypt(self.email_hash.encode('utf-8')) 是可以的，但是，因为 email是一个hybrid_property，当SQLAlchemy用它生成SQL时self.email_hash实际上是一个sqlalchemy.orm.attributes.InstrumentedAttribute类型。
来自 docs 关于混合属性：
In many cases, the construction of an in-Python function and a
SQLAlchemy SQL expression have enough differences that two separate
Python expressions should be defined.
因此您可以定义一个 hybrid_property.expression 方法，SQLAlchemy 将使用该方法生成 sql，从而使您在 hybrid_property 中保持字符串处理完好无损]方法。
根据您的示例，这是我最终得到的对我有用的代码。为简单起见，我从您的 User 模型中删除了很多内容，但所有重要部分都在那里。我还必须为您的代码中调用但未提供的其他 functions/classes 编写实现（请参阅 MCVE）：
class Fernet:
def __init__(self, k):
self.k = k
def encrypt(self, s):
return s
def decrypt(self, s):
return s
def get_env_variable(s):
return s
def generate_password_hash(s):
return s
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email_hash = db.Column(db.String(184), unique=True, nullable=False)
f = Fernet(get_env_variable('FERNET_KEY'))
@hybrid_property
def email(self):
return self.f.decrypt(self.email_hash.encode('utf-8'))
@email.expression
def email(cls):
return cls.f.decrypt(cls.email_hash)
@email.setter
def email(self, email):
self.email_hash = self.f.encrypt(email.encode('utf-8'))
if __name__ == '__main__':
db.drop_all()
db.create_all()
u = User(email='a@example.com')
db.session.add(u)
db.session.commit()
print(User.query.filter_by(email='a@example.com').first())
# <User 1>
不幸的是，上面的代码之所以有效，是因为 mock Fernet.decrypt 方法 returns 是传入的确切对象。存储用户电子邮件地址的 Fernet 编码散列的问题在于Fernet.encrypt 不会 return 从一次执行到下一次执行相同的 fernet token，即使使用相同的密钥。例如：
>>> from cryptography.fernet import Fernet
>>> f = Fernet(Fernet.generate_key())
>>> f.encrypt('a@example.com'.encode('utf-8')) == f.encrypt('a@example.com'.encode('utf-8'))
False
因此，您想查询数据库中的一条记录，但无法知道您正在查询的字段在查询时实际存储的值是多少。您可以构建一个 classmethod 来查询整个 users table 并循环遍历每条记录，解密其存储的哈希并将其与明文电子邮件进行比较。或者您可以构建一个始终 return 相同值的哈希函数，使用该函数对新用户电子邮件进行哈希处理，并直接使用电子邮件字符串的哈希值查询 email_hash 字段。其中，考虑到大量用户，第一个效率非常低。
Fernet.encrypt函数是：
def encrypt(self, data):
current_time = int(time.time())
iv = os.urandom(16)
return self._encrypt_from_parts(data, current_time, iv)
所以，你可以定义current_time和iv的静态值，然后自己直接调用Fermat._encrypt_from_parts。或者您可以使用 python 内置于 hash 和 just set a fixed seed 中，以便它是确定性的。然后，您可以散列要查询的电子邮件字符串，并首先直接查询 Users.email_hash。只要您没有对密码字段执行上述任何操作！
我在 SQLAlchemy 中有一个用户 class。我希望能够在数据库中加密用户的电子邮件地址属性，但仍然可以通过过滤器查询对其进行搜索。
我的问题是，如果我使用@hybrid_property，我的查询理论上可以工作，但我的构造没有，如果我使用@属性，我的构造可以工作，但我的查询没有
from cryptography.fernet import Fernet # <- pip install cryptography
from werkzeug.security import generate_password_hash
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email_hash = db.Column(db.String(184), unique=True, nullable=False)
password_hash = db.Column(db.String(128))
# @property # <- Consider this as option 2...
@hybrid_property # <- Consider this as option 1...
def email(self):
f = Fernet('SOME_ENC_KEY')
value = f.decrypt(self.email_hash.encode('utf-8'))
return value
@email.setter
def email(self, email):
f = Fernet('SOME_ENC_KEY')
self.email_hash = f.encrypt(email.encode('utf-8'))
@property
def password(self):
raise AttributeError('password is not a readable attribute.')
@password.setter
def password(self, password):
self.password_hash = generate_password_hash(password)
def __init__(self, **kwargs):
super(User, self).__init__(**kwargs)
# other checks and modifiers
对于选项 1：当我尝试使用 User(email='a@example.com',password='secret') 构建用户时，我收到回溯，
~/models.py in __init__(self, **kwargs)
431 # Established role assignment by default class initiation
432 def __init__(self, **kwargs):
--> 433 super(User, self).__init__(**kwargs)
434 if self.role is None:
435 _default_role = Role.query.filter_by(default=True).first()
~/lib/python3.6/site-packages/sqlalchemy/ext/declarative/base.py in _declarative_constructor(self, **kwargs)
697 raise TypeError(
698 "%r is an invalid keyword argument for %s" %
--> 699 (k, cls_.__name__))
700 setattr(self, k, kwargs[k])
701 _declarative_constructor.__name__ = '__init__'
TypeError: 'email' is an invalid keyword argument for User
对于选项 2：如果我改为将 @hybrid_property 更改为 @属性 构造没问题，但是我的查询 User.query.filter_by(email=form.email.data.lower()).first() 失败并且 returns None.
我应该更改什么才能使其按要求工作？
==============
请注意，我应该说我已经尽量避免使用双重属性，因为我不想对底层代码库进行大量编辑。所以我明确地试图避免根据 User(email_input='a@a.com', password='secret') 和 User.query.filter_by(email='a@a.com').first():
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email_hash = db.Column(db.String(184), unique=True, nullable=False)
password_hash = db.Column(db.String(128))
@hybrid_property
def email(self):
f = Fernet('SOME_ENC_KEY')
value = f.decrypt(self.email_hash.encode('utf-8'))
return value
@property
def email_input(self):
raise AttributeError('email_input is not a readable attribute.')
@email_input.setter
def email_input(self, email):
f = Fernet('SOME_ENC_KEY')
self.email_hash = f.encrypt(email.encode('utf-8'))
@property
def password(self):
raise AttributeError('password is not a readable attribute.')
@password.setter
def password(self, password):
self.password_hash = generate_password_hash(password)
def __init__(self, **kwargs):
super(User, self).__init__(**kwargs)
# other checks and modifiers
在您的 hybrid_property、email 中，如果 self.email_hash 是 str 类型，则行 self.f.decrypt(self.email_hash.encode('utf-8')) 是可以的，但是，因为 email是一个hybrid_property，当SQLAlchemy用它生成SQL时self.email_hash实际上是一个sqlalchemy.orm.attributes.InstrumentedAttribute类型。
来自 docs 关于混合属性：
In many cases, the construction of an in-Python function and a SQLAlchemy SQL expression have enough differences that two separate Python expressions should be defined.
因此您可以定义一个 hybrid_property.expression 方法，SQLAlchemy 将使用该方法生成 sql，从而使您在 hybrid_property 中保持字符串处理完好无损]方法。
根据您的示例，这是我最终得到的对我有用的代码。为简单起见，我从您的 User 模型中删除了很多内容，但所有重要部分都在那里。我还必须为您的代码中调用但未提供的其他 functions/classes 编写实现（请参阅 MCVE）：
class Fernet:
def __init__(self, k):
self.k = k
def encrypt(self, s):
return s
def decrypt(self, s):
return s
def get_env_variable(s):
return s
def generate_password_hash(s):
return s
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email_hash = db.Column(db.String(184), unique=True, nullable=False)
f = Fernet(get_env_variable('FERNET_KEY'))
@hybrid_property
def email(self):
return self.f.decrypt(self.email_hash.encode('utf-8'))
@email.expression
def email(cls):
return cls.f.decrypt(cls.email_hash)
@email.setter
def email(self, email):
self.email_hash = self.f.encrypt(email.encode('utf-8'))
if __name__ == '__main__':
db.drop_all()
db.create_all()
u = User(email='a@example.com')
db.session.add(u)
db.session.commit()
print(User.query.filter_by(email='a@example.com').first())
# <User 1>
不幸的是，上面的代码之所以有效，是因为 mock Fernet.decrypt 方法 returns 是传入的确切对象。存储用户电子邮件地址的 Fernet 编码散列的问题在于Fernet.encrypt 不会 return 从一次执行到下一次执行相同的 fernet token，即使使用相同的密钥。例如：
>>> from cryptography.fernet import Fernet
>>> f = Fernet(Fernet.generate_key())
>>> f.encrypt('a@example.com'.encode('utf-8')) == f.encrypt('a@example.com'.encode('utf-8'))
False
因此，您想查询数据库中的一条记录，但无法知道您正在查询的字段在查询时实际存储的值是多少。您可以构建一个 classmethod 来查询整个 users table 并循环遍历每条记录，解密其存储的哈希并将其与明文电子邮件进行比较。或者您可以构建一个始终 return 相同值的哈希函数，使用该函数对新用户电子邮件进行哈希处理，并直接使用电子邮件字符串的哈希值查询 email_hash 字段。其中，考虑到大量用户，第一个效率非常低。
Fernet.encrypt函数是：
def encrypt(self, data):
current_time = int(time.time())
iv = os.urandom(16)
return self._encrypt_from_parts(data, current_time, iv)
所以，你可以定义current_time和iv的静态值，然后自己直接调用Fermat._encrypt_from_parts。或者您可以使用 python 内置于 hash 和 just set a fixed seed 中，以便它是确定性的。然后，您可以散列要查询的电子邮件字符串，并首先直接查询 Users.email_hash。只要您没有对密码字段执行上述任何操作！