SQLAlchemy 버전 관리는 클래스 가져오기 순서를 고려합니다.

| | | | | | | | | |

여기서 가이드를 따르고 있었습니다.

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

문제를 발견했습니다. 다음과 같이 내 관계를 정의했습니다.

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

문자열을 사용하여 "내 모델 모듈의 가져오기 순서는 신경 쓰지 않습니다. 이 모든 것이 정상적으로 작동하지만 버전 관리 메타를 사용할 때 다음 오류가 발생합니다.

sqlalchemy.exc.InvalidRequestError: When initialization mapper Mapper|MyClass|stuffs, 표현식 "Trader"가 이름을 찾지 못했습니다("이름 "MyClass"가 정의되지 않았습니다."). 이것이 클래스 이름인 경우 두 종속 클래스가 모두 정의된 후 이 관계()를 클래스에 추가하는 것을 고려하십시오. .

다음 오류를 추적했습니다.

 파일 "/home/nick/workspace/gm3/gm3/lib/history_meta.py", 줄 90, __init__에서 매퍼 = class_mapper(cls) 파일 "/home/nick/venv/tg2env/lib/python2.6/site-packages/sqlalchemy/orm/util.py", 622행, class_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 

문제를 해결했습니다. try:except 항목을 람다에 넣고 모든 가져오기가 발생한 후에 실행합니다. 이것은 작동하지만 약간 쓰레기처럼 보입니다. 이 문제를 해결하는 더 좋은 방법에 대한 아이디어가 있습니까?

감사합니다!

업데이트

문제는 실제로 수입 순서에 관한 것이 아닙니다. 버전 관리 예제는 매퍼가 버전이 지정된 각 클래스의 공동 생성자에서 컴파일을 요구하도록 설계되었습니다. 그리고 관련 클래스가 아직 정의되지 않은 경우 컴파일이 실패합니다. 순환 관계의 경우 매핑된 클래스의 정의 순서를 변경하여 작동하게 할 방법이 없습니다.

업데이트 2

위의 업데이트 상태( 여기에서 다른 사람들의 게시물을 편집할 수 있다는 것을 몰랐습니다. :)) 이것은 순환 참조 때문일 수 있습니다. 어떤 경우에 누군가가 내 해킹을 유용하다고 생각할 수 있습니다(저는 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 (매퍼) 제외 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 history [func() for func in create_mappers] 

이렇게 하면 매퍼만 생성됩니다. 모든 클래스가 정의된 후.

업데이트 3 약간 관련이 없지만 일부 상황에서 중복 기본 키 오류가 발생했습니다(한 번에 동일한 개체에 2개의 변경 사항 커밋) . 내 해결 방법은 새 기본 자동 증가 키를 추가하는 것입니다. 물론 mysql에서는 1개 이상을 가질 수 없으므로 히스토리 테이블을 생성하는 데 사용된 기존 항목의 기본 키를 제거해야 했습니다. 전체 코드(hist_id 포함 및 외래 키 제약 제거 포함)를 확인하세요.

"""sqlalchemy.ext.declarative import의 DeclarativeMeta에서 공식 sqlalchemy 사본 """ 가져오기 매퍼, class_mapper, 속성, sqlalchemy.orm.exc import의 object_mapper 가져오기 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 table(col,references 테이블 참조) : for fk in col.foreign_keys: if fk.references(table): return True return False def _history_mapper(local_mapper): cls = local_mapper.class_ # 열 매핑된 a에서 "active_history" 플래그 설정 # on 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__", 없음) polymorphic_on = 없음 super_fks = [] super_mapper 또는 local_mapper.local_table이 아닌 경우 super_mapper.local_table: cols = [] local_mapper.local_table.c의 열: if column.name == "버전": 계속 col = column.copy() col.unique = False #col.autoincrement: col.autoincrement = False인 경우 일반 db에서 항목을 자동으로 증가시키지 않습니다. col.primary_key인 경우 합성 키: col.primary_key = super_mapper 및 col_references_table(column, super_mapper.local_table)인 경우 False: super_fks.append((col.key, list(super_history_mapper.base_mapper.local_table.primary_key)[0])) 열. 열이 local_mapper.polymorphi인 경우 추가(col) c_on: polymorphic_on = col #if super_mapper: # super_fks.append(("버전", super_history_mapper.base_mapper.local_table.c.version)) cols.append(Column("hist_id", 정수, 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 = 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, 테이블, 상속=super_history_mapper, polymorphic_on=polymorphic_identity, 다형성 =local_mapper.polymorphic_identity ) cls.__history_mapper__ = super_history_mapper가 아닌 경우 m: cls.version = Column("version", Integer, default=1, nullable=False) create_mappers = [] class VersionedMeta(DeclarativeMeta): def __initname__(cls, class , 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(람다: make_mapper()) def versioned_objects(iter): 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()): hm.single: hm.single: hist_col의 계속 .local_table.c: hist_col.key == "version" 또는 hist_col.key == "changed" 또는 hist_col.key == "hist_id"인 경우: 계속 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가 아니면 obj_changed: # 변경되지 않았지만 관계가 있습니다. OK # obj_mapper.iterate_properties의 prop에 대해서도 확인하십시오. if isinstance(prop, RelationshipProperty) and attributes.get_history(obj, prop.key).has_changes(): obj_changed = obj_changed 및 삭제되지 않은 경우 True break: return attr[" version"] = obj.version hist = 키에 대한 history_cls(), attr.iteritems()의 값: setattr(hist, key, value) obj.version += 1 session.add(hist) class VersionedListener(SessionExtension): def before_flush(self, session, flush_context, instance): obj in versioned_objects(session.dirty): create_version(obj, session) 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