I'm very new to fastapi and python altogether.
I'm using pgcrypto
module to encrypt the personal info of the customers.
My raw query is
select pgp_sym_decrypt(cast(email as bytea), 'secret_key') as email
from customers
and the query works fine.
How to produce something similar in SqlAlchemy
? I've tried something like this
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'))
Also tried
customer = select(func.pgp_sym_decrypt(cast(Customer.c.email, BYTEA), 'secret_key'))
Also tried
customer = select(func.pgp_sym_decrypt(cast(Customer.c.email, LargeBinary), 'secret_key'))
But no luck anywhere. So, how to solve this problem? Any help would be appreciated. Thank you so much in advance.
The pgp_sym_decypt
function takes a BYTEA
as its argument, but pgp_sym_encrypt
takes a VARCHAR
, so casting is unnecessary.
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())