Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #!/usr/bin/env python
- import contextlib
- import sqlalchemy
- import sqlalchemy.ext.declarative
- import sqlalchemy.orm
- import sqlalchemy.exc
- exc = sqlalchemy.exc
- from sqlalchemy import (
- Column,
- ForeignKey,
- Index,
- Integer,
- Numeric,
- )
- from sqlalchemy.orm import (
- relationship,
- )
- Base = sqlalchemy.ext.declarative.declarative_base()
- #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
- class ORDER(Base):
- __tablename__ = 'ORDERS'
- ORDERSID = Column(Integer, primary_key=True)
- ORDERSCOST = Column(Numeric(19, 4), nullable=False)
- # ORDEREDITEM = relationship("ORDEREDITEM")
- ORDEREDITEM = relationship("ORDEREDITEM", cascade="all,delete")
- class ORDEREDITEM(Base):
- __tablename__ = 'ORDEREDITEMS'
- __table_args__ = (
- Index('AK_ORDERSID_ITEMID', 'ORDERSID', 'ITEMSID', unique=True),
- )
- ORDEREDITEMSID = Column(Integer, primary_key=True)
- # ITEMSID = Column(ForeignKey(u'ITEMS.ITEMSID'), nullable=False, index=True)
- ITEMSID = Column(Integer, nullable=False, index=True)
- ORDERSID = Column(ForeignKey(u'ORDERS.ORDERSID', ondelete=u'CASCADE'), nullable=False)
- ORDEREDITEMSQUANTITY = Column(Integer, nullable=False)
- ORDER = relationship(u'ORDER')
- #~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
- engine = sqlalchemy.create_engine('sqlite:///:memory:', echo=False)
- Base.metadata.create_all(engine)
- Session = sqlalchemy.orm.sessionmaker(bind=engine)
- DBSession = Session
- def raise_database_error(error):
- print('db error', error)
- def main():
- orderID = 1
- order = ORDER(
- ORDERSID = orderID,
- ORDERSCOST = 20.0,
- )
- item = ORDEREDITEM(
- ORDEREDITEMSID = 1,
- ITEMSID = 1,
- ORDERSID = orderID,
- ORDEREDITEMSQUANTITY = 2,
- )
- with contextlib.closing(DBSession()) as session:
- session.add(order)
- session.add(item)
- session.commit()
- order = session.query(ORDER).filter_by(ORDERSID=orderID).first()
- assert order is not None
- item = session.query(ORDEREDITEM).filter_by(ORDEREDITEMSID=1).first()
- assert item is not None
- with contextlib.closing(DBSession()) as session:
- try:
- returnedOrder = session.query(ORDER).filter_by(ORDERSID=orderID).first()
- session.delete(returnedOrder)
- session.commit()
- except exc.SQLAlchemyError as error:
- session.rollback()
- raise_database_error(error)
- # else:
- # return '1'
- with contextlib.closing(DBSession()) as session:
- order = session.query(ORDER).filter_by(ORDERSID=orderID).first()
- assert order is None
- item = session.query(ORDEREDITEM).filter_by(ORDEREDITEMSID=1).first()
- assert item is None
- if __name__ == '__main__':
- main()
Add Comment
Please, Sign In to add comment