为 sqlalchemy 中的 pgm_sym_decrypt 将列转换为 postgresql 中的 bytea 类型

Question

请不要关闭这个问题。这不是重复的

我对 fastapi 和 python 完全陌生。 我正在使用pgcrypto 模块来加密客户的个人信息。 我的原始查询是

select pgp_sym_decrypt(cast(email as bytea), 'secret_key') as email
from customers

并且查询工作正常。 如何在SqlAlchemy 中产生类似的东西？我已经尝试过这样的事情

from sqlalchemy import select, func, cast, LargeBinary
from sqlalchemy.dialects.postgresql import BYTEA

customer = select(func.pgp_sym_decrypt(cast(Customer.c.email, 'bytea'), 'secret_key'))

也试过了

customer = select(func.pgp_sym_decrypt(cast(Customer.c.email, BYTEA), 'secret_key'))

也试过了

customer = select(func.pgp_sym_decrypt(cast(Customer.c.email, LargeBinary), 'secret_key'))

但是任何地方都没有运气。 那么，如何解决这个问题呢？任何帮助，将不胜感激。非常感谢您。

Answer 1

pgp_sym_decypt 函数接受 BYTEA 作为其参数，但 pgp_sym_encrypt 接受 VARCHAR，因此不需要强制转换。

import sqlalchemy as sa

engine = sa.create_engine('postgresql+psycopg2:///test', echo=True, future=True)

tbl = sa.Table('t70770085', sa.MetaData(),
               sa.Column('id', sa.Integer, primary_key=True),
               sa.Column('name', sa.String))
tbl.drop(engine, checkfirst=True)
tbl.create(engine)

SECRET = 'secret_key'

with engine.begin() as conn:
    conn.execute(tbl.insert().values(name=sa.func.pgp_sym_encrypt('Alice', SECRET)))

with engine.connect() as conn:
    query = sa.select(sa.func.pgp_sym_decrypt(sa.cast(tbl.c.name, sa.LargeBinary), SECRET))
    result = conn.execute(query)
    print(result.scalar_one())