Python如何通过类变量获取所属类？微ORM仿SQLAlchemy查询机制疑问

核心原理说明 #

你示例里的A.var是普通字符串实例，Python不会为这类内置类型的值自动存储所属宿主类的引用，因此没有通用内置属性（类似方法的__qualname__）可以直接反向获取到类A。
SQLAlchemy的Column能实现归属识别，核心是它是自定义的描述符类型，会在声明式映射类被定义的阶段，主动将所属类、列名等元信息绑定到Column实例本身。

针对你的ORM实现的改造方案 #

你现在的Col类直接继承了str，无法存储额外元信息，需要调整实现，同时借助Python的__init_subclass__钩子自动绑定归属信息：

1. 改造Col类 #

class Col:
    def __init__(self, type_def):
        # 存储列的类型定义，比如INTEGER PRIMARY KEY
        self.type_def = type_def
        # 预留字段：列对应的类属性名，后续自动绑定
        self.name = None
        # 预留字段：列所属的表映射类，后续自动绑定
        self.owner = None
    
    # 保留原字符串拼接行为，方便生成SQL时直接使用
    def __str__(self):
        return f"{self.owner.__tablename__}.{self.name}"

2. 给Base类增加自动绑定逻辑 #

class Base:
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        # 子类定义时自动遍历所有Col属性，绑定归属信息
        for attr_name, attr_val in vars(cls).items():
            if isinstance(attr_val, Col):
                attr_val.name = attr_name
                attr_val.owner = cls

    @classmethod
    def childs(cls):
        subclasses = set()
        work = [cls]
        while work:
            parent = work.pop()
            for child in parent.__subclasses__():
                if child not in subclasses:
                    subclasses.add(child)
                    work.append(child)
        return subclasses

    @classmethod
    def create(cls, cursor):
        query = 'CREATE TABLE IF NOT EXISTS {} ({})'.format(
            cls.__tablename__,
            ', '.join(
                f"{k} {v.type_def}"
                for k, v in vars(cls).items()
                if isinstance(v, Col)
            )
        )
        cursor.execute(query)

    @staticmethod
    def create_all_tables(cursor):
        for cls in __class__.childs():
            cls.create(cursor)

    @classmethod
    def drop(cls, cursor):
        query = 'DROP TABLE IF EXISTS {}'.format(
            cls.__tablename__,
        )
        cursor.execute(query)

    @classmethod
    def insert(cls, cursor, kvRow):
        query = 'INSERT INTO {} ({}) VALUES(:{})'.format(
            cls.__tablename__,
            ', '.join(kvRow.keys()),
            ', :'.join(kvRow.keys())
        )
        cursor.execute(query, kvRow)

3. 效果验证 #

定义完映射类之后，直接访问列属性就能拿到所属表信息：

class User(Base):
    __tablename__ = 'user'
    id = Col('INTEGER PRIMARY KEY')
    name = Col('VARCHAR')

print(User.id.owner) # 输出<class '__main__.User'>
print(User.id.owner.__tablename__) # 输出'user'
print(User.id.name) # 输出'id'

session.query的实现逻辑 #

你自己实现query方法时，接收多个Col实例作为参数，每个实例都能通过owner属性拿到对应的映射类，自然就能识别对应的表名，拼接SQL时直接拼接表名.列名即可：

def query(*cols):
    select_part = ', '.join(str(col) for col in cols)
    # 根据列所属的owner去重得到要查询的表列表
    tables = list({col.owner.__tablename__ for col in cols})
    from_part = ', '.join(tables)
    return f"SELECT {select_part} FROM {from_part}"

# 调用示例
print(query(User.id, User.name)) 
# 输出 SELECT user.id, user.name FROM user

内容的提问来源于stack exchange，提问作者rho