0
我爲使用SQLAlchemy存儲數據的應用程序創建了一些基類。其中一個基類（內容）是多態的，並且具有一些常規字段，如id，標題，描述，時間戳等。該類的子類應該添加存儲在單獨表中的附加字段。我創建了一個獨立的代碼示例，更好地說明了這個概念。該示例包含基類，一些子類和一些引導代碼來創建sqlite數據庫。通過將代碼粘貼到'example.py'中創建一個virtualenv，將SQLAlchemy安裝到該virtualenv中，並使用它的解釋器來運行該示例，從而獲得運行示例的最簡單方法。該示例包含一些註釋麻煩的代碼，如果該代碼被評論，示例應該運行沒有錯誤（至少它在這裏）。SQLAlchemy：指向同一個表的多個ForeignKeys，有些是可選的，啓用了多態繼承
通過取消評論代碼的註釋，該示例失敗，我不太清楚如何解決這個問題 - 任何幫助都是非常棒的！
實施例的概述：
- 它有一些基類（Base和內容）。
- 它有一個擴展內容的任務類。
- 任務可能有子任務，位置排序應該持續。
- 它有一個擴展內容的項目類（註釋）。
- 項目有一個due_date和里程碑（這是一個任務列表）
- 它有一個Worklist類（註釋），它擴展了內容。
- 工作列表屬於「員工」並且有任務。
我想要實現的是將任務工作作爲一個獨立的類，但其他類也可能有任務（如項目和工作列表）。我不想結束幾個任務/相關表格，而是希望利用Content來實現這個概念，並以這種「通用」方式附加任務。
示例代碼：
from datetime import datetime
from datetime import timedelta
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import Boolean
from sqlalchemy import String
from sqlalchemy import DateTime
from sqlalchemy import Date
from sqlalchemy import Unicode
from sqlalchemy import UnicodeText
from sqlalchemy import ForeignKey
from sqlalchemy import MetaData
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.orderinglist import ordering_list
from sqlalchemy.orm import Session
from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship
from sqlalchemy.orm import backref
from sqlalchemy.util import classproperty
class Base(object):
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
@property
def columns(self):
return self.__mapper__.columns.keys()
def add(self, **data):
self.update(**data)
db_session.add(self)
db_session.flush()
def delete(self):
db_session.delete(self)
db_session.flush()
def update(self, **data):
"""
Iterate over all columns and set values from data.
"""
for attr in self.columns:
if attr in data and data[attr] is not None:
setattr(self, attr, data[attr])
engine = create_engine('sqlite:///test.db', echo=True)
metadata = MetaData()
db_session = scoped_session(sessionmaker(bind=engine))
Base = declarative_base(cls=Base)
Base.metadata = metadata
Base.query = db_session.query_property()
class Content(Base):
"""
Base class for all content. Includes basic features such as
ownership and timestamps for modification and creation.
"""
@classproperty
def __mapper_args__(cls):
return dict(
polymorphic_on='type',
polymorphic_identity=cls.__name__.lower(),
with_polymorphic='*')
id = Column(Integer(), primary_key=True)
type = Column(String(30), nullable=False)
owner = Column(Unicode(128))
title = Column(Unicode(128))
description = Column(UnicodeText())
creation_date = Column(DateTime(), nullable=False, default=datetime.utcnow)
modification_date = Column(DateTime(), nullable=False, default=datetime.utcnow)
def __init__(self, **data):
self.add(**data)
def update(self, touch=True, **data):
"""
Iterate over all columns and set values from data.
:param touch:
:param data:
:return:
"""
super(Content, self).update(**data)
if touch and 'modification_date' not in data:
self.modification_date = datetime.utcnow()
def __eq__(self, other):
return isinstance(other, Content) and self.id == other.id
def get_content(id):
return Content.query.get(id)
class Task(Content):
id = Column(Integer, ForeignKey(Content.id), primary_key=True)
# content_id = Column(Integer, ForeignKey(Content.id), nullable=True)
done = Column(Boolean, default=False)
position = Column(Integer, default=0)
parent_id = Column(Integer, ForeignKey('task.id'), nullable=True)
tasks = relationship(
'Task',
cascade='all, delete, delete-orphan',
backref=backref('parent', remote_side=id),
foreign_keys='Task.parent_id',
order_by=position,
collection_class=ordering_list('position', reorder_on_append=True)
)
def default_due_date():
return datetime.utcnow() + timedelta(days=60)
# class Project(Content):
#
# id = Column(Integer, ForeignKey(Content.id), primary_key=True)
# due_date = Column(Date, default=default_due_date)
#
# milestones = relationship(
# 'Task',
# cascade='all, delete, delete-orphan',
# backref=backref('content_parent', remote_side=id),
# foreign_keys='Task.content_id',
# collection_class=ordering_list('position', reorder_on_append=True)
# )
#
#
# class Worklist(Content):
#
# id = Column(Integer, ForeignKey(Content.id), primary_key=True)
# employee = Column(Unicode(128), nullable=False)
#
# tasks = relationship(
# 'Task',
# cascade='all, delete, delete-orphan',
# backref=backref('content_parent', remote_side=id),
# foreign_keys='Task.content_id',
# collection_class=ordering_list('position', reorder_on_append=True)
# )
def main():
db_session.registry.clear()
db_session.configure(bind=engine)
metadata.bind = engine
metadata.create_all(engine)
# Test basic operation
task = Task(title=u'Buy milk')
task = get_content(task.id)
# assert Content attributes inherited
assert task.title == u'Buy milk'
assert task.done == False
# add subtasks
task.tasks = [
Task(title=u'Remember to check expiration date'),
Task(title=u'Check bottle is not leaking')
]
# assert that subtasks is added and correctly ordered
task = get_content(task.id)
assert len(task.tasks) == 2
assert [(x.position, x.title) for x in task.tasks] == \
[(0, u'Remember to check expiration date'),
(1, u'Check bottle is not leaking')]
# reorder subtasks
task.tasks.insert(0, task.tasks.pop(1))
task = get_content(task.id)
assert len(task.tasks) == 2
assert [(x.position, x.title) for x in task.tasks] == \
[(0, u'Check bottle is not leaking'),
(1, u'Remember to check expiration date')]
# # Test Project implementation
# project = Project(title=u'My project')
# milestone1 = Task(title=u'Milestone #1', description=u'First milestone')
# milestone2 = Task(title=u'Milestone #2', description=u'Second milestone')
# milestone1.tasks = [Task(title=u'Subtask for Milestone #1'), ]
# milestone2.tasks = [Task(title=u'Subtask #1 for Milestone #2'),
# Task(title=u'Subtask #2 for Milestone #2')]
# project.milestones = [milestone1, milestone2]
# project = get_content(project.id)
# assert project.title == u'My project'
# assert len(project.milestones) == 2
# assert [(x.position, x.title) for x in project.milestones] == \
# [(0, u'Milestone #1'), (1, u'Milestone #2')]
# assert len(Task.query.all()) == 8
# assert isinstance(milestone1.content_parent, Project) == True
#
# # Test Worklist implementation
# worklist = Worklist(title=u'My worklist', employee=u'Torkel Lyng')
# worklist.tasks = [
# Task(title=u'Ask stackoverflow for help'),
# Task(title=u'Learn SQLAlchemy')
# ]
# worklist = get_content(worklist.id)
# assert worklist.title == u'My worklist'
# assert worklist.employee == u'Torkel Lyng'
# assert len(worklist.tasks) == 2
# assert len(Task.query.all()) == 10
# assert isinstance(worklist.tasks[0].content_parent, Worklist) == True
if __name__=='__main__':
main()
我這麼久例如遺憾，希望提供一些工作獨立。任何幫助，評論設計或建議都非常令人滿意。