Tortoise ORM postgresql Encryption

Question

I figured out how to decrypt fields

class _Decrypt(Function):
    def __init__(self, term: Any, key, alias: str | None = None) -> None:
        super().__init__("pgp_sym_decrypt", term, key, alias=alias)

class Decrypt(Aggregate):
    database_func = _Decrypt

Example:

await models.UserPhone.annotate(phone_decrypt=Decrypt("phone", key)).filter(phone_decrypt=phone)

Result:

DEBUG:tortoise.db_client:SELECT pgp_sym_decrypt("phone",$1) "phone_decrypt",1 FROM "users_phones" WHERE pgp_sym_decrypt("phone",$2)=$3 LIMIT $4: ['...', '...', '...', 1]

But i cant encrypt values while creating

I tried

class Encrypt(Function): # NOT WORKING
    def __init__(self, term: Any, key, alias: str | None = None) -> None:
        super().__init__("pgp_sym_encrypt", term, key, alias=alias)

await models.UserPhone.create(phone=Encrypt(phone, key))

But this generating me only string containing my function

DEBUG:tortoise.db_client:INSERT INTO "users_phones" ("id","phone") VALUES ($1,$2): ['id', "pgp_sym_encrypt('...','...')"]

After it inserts second param tortoise throws an Exception

tortoise.exceptions.OperationalError: invalid input for query argument $2: "pgp_sym_encrypt('...','... (a bytes-like object is required, not 'str')

I think this is because function is not beeing called and interpreted as str

I tried to make get_sql() function in Encrypt, but its never beeing called

__str__ <- this is called out then i pass Encrypt() as a value

function_cast() didnt work either because its being called then selecting a field I need to encrypt only then creating or updating

So do any one have ideas on how to force tortoise use function instead of converting it to str?

Thanks for your attention ♥

I have an answer, no other ways i supose :( py class Encrypt: def __init__(self, key): self.key = key async def encrypt(self, value) -> bytes: row = await connections.get("default").execute_query_dict("SELECT pgp_sym_encrypt($1, $2) AS ans;", [value, self.key]) return row[0]["ans"] — london
– london, Commented Sep 15 at 19:35
if you have answer then you can put it as Answer below. And describe it - so it could be useful for others. — furas
– furas, Commented Sep 15 at 20:08

london · Accepted Answer · 2025-10-05 08:04:16Z

0

class Encrypt:
    def __init__(self, key):
        self.key = key

    async def encrypt(self, value) -> bytes:
        row = await connections.get("default").execute_query_dict("SELECT pgp_sym_encrypt($1, $2) AS ans;", [value, self.key])
        return row[0]["ans"]

Usage:

await models.ModelWithEncryptedField.create(encrypted=await Encrypt(key).encrypt(value))

answered Oct 5 at 8:04

london

113 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

Tortoise ORM postgresql Encryption

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related