[Python] How to track access to class variables in a custom ORM using Python descriptors?

I'm building my own ORM in Python and ran into an issue. I want to track the access to class variables in order to know which ForeignKey is being used in each part of my code.

For example, consider the following chain of access with this 3 classes:

class Country(Table):
__table_name__ = "country"

country_id: Column[int] = Column(int, is_primary_key=True)
country: Column[str]
last_update: Column[datetime]


class City(Table):
__table_name__ = "city"

city_id: Column[int] = Column(int, is_primary_key=True)
city: Column[str]
country_id: Column[int]
last_update: Column[datetime]

Country = ForeignKey["City", Country](Country, lambda ci, co: ci.country_id == co.country_id)


class Address(Table):
__table_name__ = "address"

address_id: Column[int] = Column(int, is_primary_key=True)
address: Column[str]
address2: Column[str]
district: Column[str]
city_id: Column[int]
postal_code: Column[str]
phone: Column[str]
location: Column[str]
last_update: Column[datetime] = Column(datetime, is_auto_generated=True)

City = ForeignKey["Address", City](City, lambda a, c: a.city_id == c.city_id)

if __name__ == "__main__":

Address.City.Country.country


When using this syntax in different parts of my code, I’d like to extract the names of the ForeignKey attributes as a list of strings, in some way, so I can determine which tables should be included in the JOINs.

['Address', 'City', 'Country']



To provide some context, the ForeignKey class is implemented as a descriptor. My goal is to "record" each access in the chain so I can analyze the path of the foreign keys.

How can I achieve this behavior in Python? Are there specific techniques or patterns that would allow me to track this chain of attribute accesses?

An example of a complex query would be:

import unittest

from config import config_dict # noqa: E402
from ormlambda.databases.my_sql.repository import MySQLRepository # noqa: E402
from ormlambda import Table, ForeignKey, Column, BaseModel # noqa: E402
from ormlambda.common.interfaces.IRepositoryBase import IRepositoryBase # noqa: E402



class AddressModel(BaseModel[Address]):
def __new__[TRepo](cls, repository: IRepositoryBase[TRepo]):
return super().__new__(cls, Address, repository)



class TestWhere(unittest.TestCase):
def test_select_with_where_clause(self):
ddbb = MySQLRepository(**config_dict)
select = (
AddressModel(ddbb)
.order(Address.City.city_id, order_type="ASC")
.where(
(
(Address.address_id >= 60),
(Address.City.city_id <= 80),
(Address.City.Country.country_id < 100),
)
)
.select(
(
Address.address_id,
Address.City.city_id,
Address.City.Country.country_id,
),
flavour=dict,
)
)

response: tuple[dict[str, int]] = (
{"address_address_id": 60, "city_city_id": 49, "country_country_id": 80},
{"address_address_id": 88, "city_city_id": 40, "country_country_id": 60},
{"address_address_id": 98, "city_city_id": 79, "country_country_id": 99},
)
self.assertTupleEqual(response, select)


if __name__ == "__main__":
unittest.main()



I previously resolved this in earlier versions by using a combination of a lambda method and inspect.getsourcefile. However, this approach isn’t ideal because I cannot pass variables directly; instead, I have to wrap them into a dictionary. This limitation makes the solution cumbersome and less flexible than I would like.

Continue reading...