SQLAlchemy Core中的异常及事务处理样码
时间:2021-07-01 10:21:17
帮助过:31人阅读
coding=utf-8
from datetime
import datetime
from sqlalchemy
import (MetaData, Table, Column, Integer, Numeric, String, Boolean,
DateTime, ForeignKey, ForeignKey, create_engine, CheckConstraint)
from sqlalchemy
import (insert, select, update, delete, text, desc, cast, and_, or_, not_)
from sqlalchemy.sql
import func
from sqlalchemy.exc
import IntegrityError
metadata =
MetaData()
cookies = Table(
‘cookies‘, metadata,
Column(‘cookie_id‘, Integer(), primary_key=
True),
Column(‘cookie_name‘, String(50), index=
True),
Column(‘cookie_recipe_url‘, String(255
)),
Column(‘cookie_sku‘, String(55
)),
Column(‘quantity‘, Integer()),
Column(‘unit_cost‘, Numeric(12, 2
)),
CheckConstraint(‘quantity >= 0‘, name=
‘quantity_positive‘)
)
users = Table(
‘users‘, metadata,
Column(‘user_id‘, Integer(), primary_key=
True),
Column(‘username‘, String(15), nullable=False, unique=
True),
Column(‘email_address‘, String(255), nullable=
False),
Column(‘phone‘, String(20), nullable=
False),
Column(‘password‘ ,String(25), nullable=
False),
Column(‘created_on‘, DateTime(), default=
datetime.now),
Column(‘updated_on‘, DateTime(), default=datetime.now, onupdate=
datetime.now)
)
orders = Table(
‘orders‘, metadata,
Column(‘order_id‘, Integer(), primary_key=
True),
Column(‘user_id‘, ForeignKey(
‘users.user_id‘)),
Column(‘shipped‘, Boolean(), default=
False)
)
line_items = Table(
‘line_items‘, metadata,
Column(‘line_items_id‘, Integer(), primary_key=
True),
Column(‘order_id‘, ForeignKey(
‘orders.order_id‘)),
Column(‘cookie_id‘, ForeignKey(
‘cookies.cookie_id‘)),
Column(‘quantity‘, Integer()),
Column(‘extended_cost‘, Numeric(12, 2
))
)
engine = create_engine(
‘mysql+pymysql://u:p@ip:3306/cookies‘)
metadata.create_all(engine)
connection =
engine.connect()
def ship_it(order_id):
s =
select([line_items.c.cookie_id, line_items.c.quantity])
s = s.where(line_items.c.order_id ==
order_id)
transaction =
connection.begin()
cookies_to_ship =
connection.execute(s)
try:
for cookie
in cookies_to_ship:
u = update(cookies).where(cookies.c.cookie_id==
cookie.cookie_id)
u = u.values(quantity = cookies.c.quantity -
cookie.quantity)
result =
connection.execute(u)
u = update(orders).where(orders.c.order_id ==
order_id)
u = u.values(shipped=
True)
result =
connection.execute(u)
print(
"Shipped order ID: {}".format(order_id))
transaction.commit()
except IntegrityError as error:
transaction.rollback()
print(error)
‘‘‘
ins = insert(users).values(
username="cookiemon",
email_address="mon@cookie.com",
phone="111-111-1111",
password="password"
)
try:
result = connection.execute(ins)
except IntegrityError as error:
pass
ins = cookies.insert()
inventory_list = [
{
‘cookie_name‘: ‘chocolate chip‘,
‘cookie_recipe_url‘: ‘http://some.aweso.me/cookie/recipe.html‘,
‘cookie_sku‘: ‘CC01‘,
‘quantity‘: ‘12‘,
‘unit_cost‘: ‘0.50‘
},
{
‘cookie_name‘: ‘dark chocolate chip‘,
‘cookie_recipe_url‘: ‘http://some.aweso.me/cookie/recipe_dark.html‘,
‘cookie_sku‘: ‘CC02‘,
‘quantity‘: ‘1‘,
‘unit_cost‘: ‘0.75‘
}
]
result = connection.execute(ins, inventory_list)
ins = insert(orders).values(user_id=1, order_id=‘1‘)
result = connection.execute(ins)
ins = insert(line_items)
order_items = [
{
‘order_id‘: 1,
‘cookie_id‘: 1,
‘quantity‘: 9,
‘extended_cost‘: 4.50
}
]
result = connection.execute(ins, order_items)
ins = insert(orders).values(user_id=1, order_id=‘2‘)
result = connection.execute(ins)
ins = insert(line_items)
order_items = [
{
‘order_id‘: 2,
‘cookie_id‘: 1,
‘quantity‘: 4,
‘extended_cost‘: 1.50
},
{
‘order_id‘: 2,
‘cookie_id‘: 2,
‘quantity‘: 1,
‘extended_cost‘: 4.50
}
]
result = connection.execute(ins, order_items)
ship_it(1)
s = select([cookies.c.cookie_name, cookies.c.quantity])
connection.execute(s).fetchall()
‘‘‘
ship_it(20)
SQLAlchemy Core中的异常及事务处理样码
标签:rom username try mysq pymysql ack create created mit