0
SQLAlchemy를 사용하여 데이터를 저장하는 응용 프로그램의 기본 클래스를 만들었습니다. Base 클래스 (Content) 중 하나는 다형성이며 id, title, description, timestamp 등의 일반 필드가 있습니다.이 클래스의 서브 클래스는 별도의 테이블에 저장되는 추가 필드를 추가해야합니다. 개념을 더 잘 보여주는 독립 실행 형 코드 샘플을 만들었습니다. 이 예제에는 sqlite 데이터베이스를 만들기위한 Base 클래스, 일부 하위 클래스 및 일부 부트 스트랩 코드가 포함되어 있습니다. 'example.py'에 코드를 붙여 넣고, virtualenv를 만들고, SQLAlchemy를 virtualenv에 설치하고, 예제를 실행하기 위해 인터프리터를 사용하여 예제를 실행하는 가장 쉬운 방법입니다. 이 예제에는 주석이 달린 귀찮은 코드가 포함되어 있습니다. 코드에 주석이 달린 경우 예제가 오류없이 실행되어야합니다 (여기까지만 수행됩니다).SQLAlchemy : 동일한 테이블을 가리키는 여러 ForeignKeys 중 일부는 다형성 상속을 사용할 수있는 선택 사항입니다.
주석 처리 된 코드의 주석 처리를 제거하면 예제가 실패하고이 문제를 해결하는 방법이 확실하지 않습니다. 어떤 도움이 초능력입니다!
예 개요 :
- 그것은 몇 가지 기본 클래스 (자료 및 내용)가 있습니다.
- 콘텐츠를 확장하는 Task 클래스가 있습니다.
- 작업에 하위 작업이 포함될 수 있으므로 위치 순서는 유지되어야합니다.
- 콘텐츠를 확장하는 Project 클래스 (주석 처리 됨)가 있습니다.
- 프로젝트에는 due_date 및 중요 시점 (작업 목록)이 있습니다.
- 콘텐츠를 확장하는 Worklist 클래스 (주석 처리 됨)가 있습니다.
- 작업 목록은 '직원'에 속하고 작업을 가지고 있습니다.
내가 수행하고자하는 작업은 독립 실행 형 클래스로 작업하는 것이지만 추가 클래스에는 작업 (예 : 프로젝트 및 작업 목록)이있을 수도 있습니다. 나는 몇 가지 작업/관련 테이블로 끝내고 싶지 않지만 오히려 컨셉을 위해 컨텐트를 활용하고이 '일반적인'방식으로 작업을 첨부하려고합니다.
예제 코드 :이 긴 예를 들어 미안 해요
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()
는 독립을했다 무언가를 제공하고 싶었다. 어떤 도움, 디자인이나 제안에 대한 의견은 크게 appretiated 있습니다.