versionamento do SQLAlchemy se preocupa com a ordem de importação da classe

| | | | | | | | | |

Eu estava seguindo o guia aqui:

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

e encontrou um problema. Eu defini meus relacionamentos como:

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

com strings para que não "Não me importo com a ordem de importação dos módulos do meu modelo. Isso tudo funciona normalmente, mas quando uso o meta de versionamento recebo o seguinte erro:

sqlalchemy.exc.InvalidRequestError: When initializing mapper Mapper|MyClass|stuffs, expressão "Trader" falhou ao localizar um nome ("nome "MyClass" não está definido"). Se este for um nome de classe, considere adicionar este relacionamento() à classe depois que ambas as classes dependentes tiverem sido definidas .

Rastreei o erro para:

 Arquivo "/home/nick/workspace/gm3/gm3/lib/history_meta.py", linha 90, em __init__ mapper = class_mapper(cls) Arquivo "/home/nick/venv/tg2env/lib/python2.6/site-packages/sqlalchemy/orm/util.py", linha 622, em 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) exceto UnmappedClassError: pass 

Eu consertei o problema colocando o try: except stuff em um lambda e executando todos eles após todas as importações terem acontecido. Isso funciona, mas parece um pouco inútil, alguma idéia de como corrigir isso é uma maneira melhor?

Obrigado!

Atualização

O problema não é realmente sobre o pedido de importação. O exemplo de controle de versão foi projetado para que o mapeador exija a compilação no costructor de cada classe com versão. E a compilação falha quando as classes relacionadas ainda não estão definidas. No caso de relações circulares, não há como fazê-lo funcionar alterando a ordem de definição das classes mapeadas.

Atualização 2

Como a atualização acima afirma ( Eu não sabia que você poderia editar as postagens de outras pessoas aqui :)) isso provavelmente se deve a referências circulares. Nesse caso, alguém achará meu hack útil (estou usando-o com turbogears) (substitua VersionedMeta e adicione create_mappers global em history_meta)

create_mappers = [] class VersionedMeta(DeclarativeMeta) : def __init__(cls, classname, bases, dict_): DeclarativeMeta.__init__(cls, classname, bases, dict_) #Adicionei este código porque estava travando caso contrário def make_mapper(): try: mapper = class_mapper(cls) _history_mapper (mapper) exceto UnmappedClassError: pass create_mappers.append(lambda: make_mapper()) 

Então você pode fazer algo como o seguinte em seus modelos __init__.py

# Importe seus módulos de modelo aqui. 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] 

Dessa forma ele cria apenas os mappers depois que todas as classes foram definidas.

Atualização 3 Ligeiramente não relacionado, mas me deparei com um erro de chave primária duplicado em algumas circunstâncias (confirmando 2 alterações no mesmo objeto de uma só vez) . Minha solução alternativa foi adicionar uma nova chave primária de incremento automático. É claro que você não pode ter mais de 1 com o mysql, então eu tive que remover a chave primária do material existente usado para criar a tabela de histórico. Confira meu código geral (incluindo um hist_id e se livrar da restrição de chave estrangeira):

"""Roubado das receitas oficiais do sqlalchemy """ de sqlalchemy.ext.declarative import DeclarativeMeta de sqlalchemy.orm import mapper, class_mapper, atributos, object_mapper de sqlalchemy.orm.exc import UnmappedClassError, UnmappedColumnError de sqlalchemy import Table, Column, ForeignKeyConstraint, Integer de sqlalchemy.orm.interfaces import SessionExtension de sqlalchemy.orm.properties import RelationshipProperty de sqlalchemy.types import DateTime import datetime de 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_ # define o sinalizador "active_history" # em uma coluna mapeada ttributes para que a versão antiga # da informação seja sempre carregada (atualmente a define em todos os atributos) para prop em local_mapper.iterate_properties: 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 = Nenhum super_fks = [] se não for super_mapper ou local_mapper.local_table não for super_mapper.local_table: cols = [] for column in local_mapper.local_table.c: if column.name == "version": continue col = column.copy() col.unique = False #não incremente automaticamente coisas do banco de dados normal se col.autoincrement: col.autoincrement = False #sqllite cai com chaves de auto incremento se tivermos um chave composta se col.primary_key: col.primary_key = False se super_mapper e 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. append(col) se a coluna for local_mapper.polymorphi 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: # herança de tabela única. pegue quaisquer colunas adicionais que possam ter sido # adicionadas e adicione-as à tabela de histórico. para coluna em local_mapper.local_table.c: se column.key não estiver em super_history_mapper.local_table.c: col = column.copy() super_history_mapper.local_table.append_column(col) table = Nenhum se 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 se não 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_) #Adicionei este código porque estava travando, caso contrário def make_mapper(): try: mapper = class_mapper(cls) _history_mapper(mapper) exceto 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, delete = False): obj_mapper = object_mapper(obj) history_mapper = obj.__history_mapper__ history_cls = history_mapper.class_ obj_state = attribute.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: continue para hist_col em hm .local_table.c: if hist_col.key == "versão" ou hist_col.key == "alterado" ou hist_col.key == "hist_id": continue obj_col = om.local_table.c[hist_col.key] # obtém o valor do atributo # baseado na MapperProperty relacionada à # coluna mapeada. isso permitirá o uso de MapperProperties # que tenha um nome de chave diferente daquele da coluna mapeada. try: prop = obj_mapper.get_property_by_column(obj_col) exceto UnmappedColumnError: # no caso de herança de tabela única, pode haver # colunas na tabela mapeada destinada apenas à subclasse. # o status "não mapeado" da coluna de subclasse na # classe base é um recurso do módulo declarativo a partir do sqla 0.5.2. continue # atributos de objetos expirados e também cols adiados podem não estar no # dict. force-o a carregar, não importa o quê, usando getattr(). se prop.key não estiver em obj_state.dict: getattr(obj, prop.key) a, u, d = attribute.get_history(obj, prop.key) se d: attr[hist_col.key] = d[0] obj_changed = True elif u: attr[hist_col.key] = u[0] else: # se o atributo não tiver valor. attr[hist_col.key] = a[0] obj_changed = Verdadeiro se não obj_changed: # não alterado, mas temos relacionamentos. OK # verifique esses também para prop em obj_mapper.iterate_properties: if isinstance(prop, RelationshipProperty) e attribute.get_history(obj, prop.key).has_changes(): obj_changed = True break if not obj_changed and not delete: 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): para obj em versioned_objects(session.dirty): create_version(obj, session) para obj em versioned_objects(session.deleted): create_version(obj, session, delete = 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