SQLAlchemy 版本控制關心類導入順序

| | | | | | | | | |

我在這裡遵循指南:

http ://www.sqlalchemy.org/docs/orm/examples.html?highlight=versioning#versioned-objects

遇到了一個問題。我已經定義了我的關係,例如:

generic_ticker = relationship("MyClass", backref=backref("stuffs")) 

with strings 所以它沒有“不關心我的模型模塊的導入順序。這一切正常,但是當我使用版本控制元時,我收到以下錯誤:

sqlalchemy.exc.InvalidRequestError:初始化映射器時Mapper|MyClass|stuffs,表達式“Trader”未能找到名稱(“名稱“MyClass”未定義”)。如果這是一個類名,請考慮在定義了兩個依賴類後將此 relationship() 添加到類.

我將錯誤追踪到:

 File "/home/nick/workspace/gm3/gm3/lib/history_meta.py", line 90,在 __init__ mapper = class_mapper(cls) 文件“/home/nick/venv/tg2env/lib/python2.6/site-packages/sqlalchemy/orm/util.py”,第 622 行,在 class_mapper mapper = mapper.compile () 

class VersionedMeta(DeclarativeMeta): def __init__(cls, cl assname, bases, dict_): DeclarativeMeta.__init__(cls, classname, bases, dict_) try: mapper = class_mapper(cls) _history_mapper(mapper) except UnmappedClassError: pass 

我解決了這個問題通過嘗試:除了 lambda 中的東西並在所有導入發生後運行它們。這可行,但似乎有點垃圾,關於如何解決這個問題的任何想法是更好的方法?

謝謝!

更新

問題實際上不在於進口訂單。版本控制示例的設計使得映射器需要在每個版本控制類的構造函數中進行編譯。當相關類尚未定義時,編譯會失敗。在循環關係的情況下,無法通過更改映射類的定義順序來使其工作。

更新 2

如上述更新所述(我不知道你可以在這裡編輯其他人的帖子 :)) 這可能是由於循環引用。在這種情況下,可能有人會發現我的 hack 很有用(我將它與 turbogears 一起使用)(替換 VersionedMeta 並在 history_meta 中添加 create_mappers global)

create_mappers = [] class VersionedMeta(DeclarativeMeta) : def __init__(cls, classname, bases, dict_): DeclarativeMeta.__init__(cls, classname, bases, dict_) #我添加了這段代碼,否則它會崩潰 def make_mapper(): try: mapper = class_mapper(cls) _history_mapper (mapper) except UnmappedClassError: pass create_mappers.append(lambda: make_mapper()) 

然後你可以在你的模型中做類似下面的事情__init__.py

<代碼># 在這裡導入你的模型模塊 from myproj.lib.history_meta import create_mappers from myproj.model.misc import * from myproj.model.actor import * from myproj.model.stuff1 import * from myproj.model.instrument import * from myproj.model.stuff import * #setup the history [func() for func in create_mappers] 

這樣它只創建映射器在定義了所有類之後。

更新 3 有點不相關,但在某些情況下我遇到了重複的主鍵錯誤(一次對同一個對象進行 2 次更改) .我的解決方法是添加一個新的主自動遞增鍵。當然你不能有超過 1 的 mysql 所以我不得不取消用於創建歷史表的現有內容的主鍵。查看我的整體代碼(包括 hist_id 和擺脫外鍵約束):

"""從官方 sqlalchemy recpies 中竊取 """ from sqlalchemy.ext.declarative import DeclarativeMeta from sqlalchemy.orm import mapper, class_mapper, attributes, object_mapper from sqlalchemy.orm.exc import UnmappedClassError, UnmappedColumnError from sqlalchemy import Table, Column, ForeignKeyConstraint, Integer from sqlalchemy.orm.interfaces import SessionExtension from sqlalchemy.orm.properties import RelationshipProperty from sqlalchemy.types import DateTime import datetime from sqlalchemy.orm.session import Session def col_references_table(col, table) : for fk in col.foreign_keys: if fk.references(table): return True return False def _history_mapper(local_mapper): cls = local_mapper.class_ # 設置“active_history”標誌 # on on column-mapped a ttributes 以便始終為 local_mapper.iterate_properties 中的 prop 加載舊版本的信息(當前在所有屬性上設置): getattr(local_mapper.class_, prop.key).impl.active_history = True super_mapper = local_mapper.inherits super_history_mapper = getattr(cls, "__history_mapper__", None) polymorphic_on = None super_fks = [] 如果不是 super_mapper 或 local_mapper.local_table 不是 super_mapper.local_table: cols = [] for column in local_mapper.local_table.c: if column.name == "version": continue col = column.copy() col.unique = False #don"t auto increment stuff from the normal db if col.autoincrement: col.autoincrement = False #sqllite 如果我們有複合鍵 if col.primary_key: col.primary_key = False if super_mapper and col_references_table(column, super_mapper.local_table): super_fks.append((col.key, list(super_history_mapper.base_mapper.local_table.primary_key)[0])) cols。如果列是 local_mapper.polymorphi,則追加(col) c_on: polymorphic_on = col #if super_mapper: # super_fks.append(("version", super_history_mapper.base_mapper.local_table.c.version)) cols.append(Column("hist_id", Integer, primary_key=True, autoincrement=True) ) cols.append(Column("version", Integer)) cols.append(Column("changed", DateTime, default=datetime.datetime.now)) if super_fks: cols.append(ForeignKeyConstraint(*zip(*super_fks) )) table = Table(local_mapper.local_table.name + "_history", local_mapper.local_table.metadata, *cols, mysql_engine="InnoDB") else: # 單表繼承。獲取可能已添加的任何其他列並將它們添加到歷史記錄表中。對於 local_mapper.local_table.c 中的列:如果 column.key 不在 super_history_mapper.local_table.c 中:col = column.copy() super_history_mapper.local_table.append_column(col) table = None 如果 super_history_mapper: bases = (super_history_mapper.class_,) else: bases = local_mapper.base_mapper.class_.__bases__ versioned_cls = type.__new__(type, "%sHistory" % cls.__name__, bases, {}) m = mapper( versioned_cls, table, inherits=super_history_mapper, polymorphic_on=polymorphic_on, polymorphic_identity =local_mapper.polymorphic_identity ) cls.__history_mapper__ = m 如果不是 super_history_mapper: cls.version = Column("version", Integer, default=1, nullable=False) create_mappers = [] class VersionedMeta(DeclarativeMeta): def __init__(cls, classname , bases, dict_): DeclarativeMeta.__init__(cls, classname, bases, dict_) #我添加了這段代碼,否則它會崩潰 def make_mapper(): try: mapper = class_mapper(cls) _history_mapper(mapper) except UnmappedClassError: pass create_mappers.append(lambda: make_mapper()) def versioned_objects(iter): for obj in iter: if hasattr(obj, "__history_mapper__"): yield obj def create_version(obj, session, deleted = False): obj_mapper = object_mapper(obj) history_mapper = obj.__history_mapper__ history_cls = history_mapper.class_ obj_state = attributes.instance_state(obj) attr = {} obj_changed = False for om, hm in zip(obj_mapper.iterate_to_root(), history_mapper.iterate_to_root()): if hm.single: 繼續 hist_col in hm .local_table.c: if hist_col.key == "version" or hist_col.key == "changed" or hist_col.key == "hist_id": continue obj_col = om.local_table.c[hist_col.key] # 獲取值基於與 # 映射列相關的 MapperProperty 的 # 屬性。這將允許使用與映射列的鍵名不同的 MapperProperties #。 try: prop = obj_mapper.get_property_by_column(obj_col) except UnmappedColumnError: # 在單表繼承的情況下,映射表上可能有 # 列僅用於子類。 # 基類上子類列的“未映射”狀態是從 sqla 0.5.2 開始的聲明性模塊的一個特性。繼續 # 過期的對象屬性和延遲的列可能不在 # 字典中。無論如何使用 getattr() 強制它加載。如果 prop.key 不在 obj_state.dict 中:getattr(obj, prop.key) a, u, d = attributes.get_history(obj, prop.key) if d: attr[hist_col.key] = d[0] obj_changed = True elif u: attr[hist_col.key] = u[0] else: # 如果屬性沒有值。 attr[hist_col.key] = a[0] obj_changed = True if not obj_changed: # 沒有改變,但我們有關係。 OK # 檢查 obj_mapper.iterate_properties 中的 prop:if isinstance(prop, RelationshipProperty) 和 attributes.get_history(obj, prop.key).has_changes(): obj_changed = True break if not obj_changed and not deleted: return attr[" version"] = obj.version hist = history_cls() for key, value in attr.iteritems(): setattr(hist, key, value) obj.version += 1 session.add(hist) class VersionedListener(SessionExtension): def before_flush(self, session, flush_context, instances): for obj in versioned_objects(session.dirty): create_version(obj, session) for obj in versioned_objects(session.deleted): create_version(obj, session, deleted = True) 

Shop

Learn programming in R: courses

$

Best Python online courses for 2022

$

Best laptop for Fortnite

$

Best laptop for Excel

$

Best laptop for Solidworks

$

Best laptop for Roblox

$

Best computer for crypto mining

$

Best laptop for Sims 4

$

Latest questions

NUMPYNUMPY

psycopg2: insert multiple rows with one query

12 answers

NUMPYNUMPY

How to convert Nonetype to int or string?

12 answers

NUMPYNUMPY

How to specify multiple return types using type-hints

12 answers

NUMPYNUMPY

Javascript Error: IPython is not defined in JupyterLab

12 answers

News


Wiki

Python OpenCV | cv2.putText () method

numpy.arctan2 () in Python

Python | os.path.realpath () method

Python OpenCV | cv2.circle () method

Python OpenCV cv2.cvtColor () method

Python - Move item to the end of the list

time.perf_counter () function in Python

Check if one list is a subset of another in Python

Python os.path.join () method