Python操作 RabbitMQ、Redis、Memcache、SQLAlchemy
來自： http://my.oschina.net/eddylinux/blog/607074
Memcached
Memcached 是一個高性能的分布式內存對象緩存系統，用于動態Web應用以減輕數據庫負載。它通過在內存中緩存數據和對象來減少讀取數據庫的次數，從而提高動態、數據庫驅動網站的速度。Memcached基于一個存儲鍵/值對的hashmap。其守護進程（daemon ）是用C寫的，但是客戶端可以用任何語言來編寫，并通過memcached協議與守護進程通信。
Memcached安裝和基本使用
Memcached安裝：
wget http://memcached.org/latest
tar -zxvf memcached-1.x.x.tar.gz
cd memcached-1.x.x
./configure && make && make test && sudo make install
PS：依賴libevent
yum install libevent-devel
apt-get install libevent-dev
啟動Memcached
memcached -d -m 10 -u root -l 10.211.55.4 -p 12000 -c 256 -P /tmp/memcached.pid
參數說明:
-d 是啟動一個守護進程
-m 是分配給Memcache使用的內存數量，單位是MB
-u 是運行Memcache的用戶
-l 是監聽的服務器IP地址
-p 是設置Memcache監聽的端口,最好是1024以上的端口
-c 選項是最大運行的并發連接數，默認是1024，按照你服務器的負載量來設定
-P 是設置保存Memcache的pid文件
Memcached命令
存儲命令: set/add/replace/append/prepend/cas
獲取命令: get/gets
其他命令: delete/stats..
Python操作Memcached
安裝API
python操作Memcached使用Python-memcached模塊
下載安裝：https://pypi.python.org/pypi/python-memcached
1、第一次操作
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
mc.set("foo", "bar")
ret = mc.get('foo')
print ret
Ps：debug = True 表示運行出現錯誤時，現實錯誤信息，上線后移除該參數。
2、天生支持集群
python-memcached模塊原生支持集群操作，其原理是在內存維護一個主機列表，且集群中主機的權重值和主機在列表中重復出現的次數成正比
主機 權重
1.1.1.1 1
1.1.1.2 2
1.1.1.3 1
那么在內存中主機列表為：
host_list = ["1.1.1.1", "1.1.1.2", "1.1.1.2", "1.1.1.3", ]
如果用戶根據如果要在內存中創建一個鍵值對（如：k1 = "v1"），那么要執行一下步驟：
根據算法將 k1 轉換成一個數字
將數字和主機列表長度求余數，得到一個值 N（ 0 <= N < 列表長度 ）
在主機列表中根據 第2步得到的值為索引獲取主機，例如：host_list[N]
連接 將第3步中獲取的主機，將 k1 = "v1" 放置在該服務器的內存中
代碼實現如下：
mc = memcache.Client([('1.1.1.1:12000', 1), ('1.1.1.2:12000', 2), ('1.1.1.3:12000', 1)], debug=True)
mc.set('k1', 'v1')
3、add
添加一條鍵值對，如果已經存在的 key，重復執行add操作異常
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
mc.add('k1', 'v1')
# mc.add('k1', 'v2') # 報錯，對已經存在的key重復添加，失敗！！！
4、replace
replace 修改某個key的值，如果key不存在，則異常
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
# 如果memcache中存在kkkk，則替換成功，否則一場
mc.replace('kkkk','999')
5、set 和 set_multi
set 設置一個鍵值對，如果key不存在，則創建，如果key存在，則修改
set_multi 設置多個鍵值對，如果key不存在，則創建，如果key存在，則修改
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
mc.set('key0', 'eddy')
mc.set_multi({'key1': 'val1', 'key2': 'val2'})
6、delete 和 delete_multi
delete 在Memcached中刪除指定的一個鍵值對
delete_multi 在Memcached中刪除指定的多個鍵值對
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
mc.delete('key0')
mc.delete_multi(['key1', 'key2'])
7、get 和 get_multi
get 獲取一個鍵值對
get_multi 獲取多一個鍵值對
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
val = mc.get('key0')
item_dict = mc.get_multi(["key1", "key2", "key3"])
8、append 和 prepend
append 修改指定key的值，在該值 后面 追加內容
prepend 修改指定key的值，在該值 前面 插入內容
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
# k1 = "v1"
mc.append('k1', 'after')
# k1 = "v1after"
mc.prepend('k1', 'before')
# k1 = "beforev1after"
9、decr 和 incr
incr 自增，將Memcached中的某一個值增加 N （ N默認為1 ）
decr 自減，將Memcached中的某一個值減少 N （ N默認為1 ）
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True)
mc.set('k1', '777')
mc.incr('k1')
# k1 = 778
mc.incr('k1', 10)
# k1 = 788
mc.decr('k1')
# k1 = 787
mc.decr('k1', 10)
# k1 = 777
10、gets 和 cas
如商城商品剩余個數，假設改值保存在memcache中，product_count = 900
A用戶刷新頁面從memcache中讀取到product_count = 900
B用戶刷新頁面從memcache中讀取到product_count = 900
如果A、B用戶均購買商品
A用戶修改商品剩余個數 product_count＝899
B用戶修改商品剩余個數 product_count＝899
如此一來緩存內的數據便不在正確，兩個用戶購買商品后，商品剩余還是 899
如果使用python的set和get來操作以上過程，那么程序就會如上述所示情況！
如果想要避免此情況的發生，只要使用 gets 和 cas 即可，如：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import memcache
mc = memcache.Client(['10.211.55.4:12000'], debug=True, cache_cas=True)
v = mc.gets('product_count')
# ...
# 如果有人在gets之后和cas之前修改了product_count，那么，下面的設置將會執行失敗，剖出異常，從而避免非正常數據的產生
mc.cas('product_count', "899")
Ps：本質上每次執行gets時，會從memcache中獲取一個自增的數字，通過cas去修改gets的值時，會攜帶之前獲取的自增值和memcache中的自增值進行比較，如果相等，則可以提交，如果不想等，那表示在gets和cas執行之間，又有其他人執行了gets（獲取了緩沖的指定值）， 如此一來有可能出現非正常數據，則不允許修改。
Memcached 真的過時了嗎？
Redis
redis是一個key-value存儲系統。和Memcached類似，它支持存儲的value類型相對更多，包括string(字符串)、list(鏈表)、set(集合)、zset(sorted set --有序集合)和hash（哈希類型）。這些數據類型都支持push/pop、add/remove及取交集并集和差集及更豐富的操作，而且這些操作都是原子性的。在此基礎上，redis支持各種不同方式的排序。與memcached一樣，為了保證效率，數據都是緩存在內存中。區別的是redis會周期性的把更新的數據寫入磁盤或者把修改操作寫入追加的記錄文件，并且在此基礎上實現了master-slave(主從)同步。
Redis安裝和基本使用
wget http://download.redis.io/releases/redis-3.0.6.tar.gz
tar xzf redis-3.0.6.tar.gz
cd redis-3.0.6
make
啟動服務端
src/redis-server
啟動客戶端
src/redis-cli
redis> set foo bar
OK
redis> get foo
"bar"
Python操作Redis
安裝API
sudo pip install redis
or
sudo easy_install redis
or
源碼安裝
詳見：
常用操作
1、操作模式
redis-py提供兩個類Redis和StrictRedis用于實現Redis的命令，StrictRedis用于實現大部分官方的命令，并使用官方的語法和命令，Redis是StrictRedis的子類，用于向后兼容舊版本的redis-py。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import redis
r = redis.Redis(host='10.211.55.4', port=6379)
r.set('foo', 'Bar')
print r.get('foo')
2、連接池
redis-py使用connection pool來管理對一個redis server的所有連接，避免每次建立、釋放連接的開銷。默認，每個Redis實例都會維護一個自己的連接池。可以直接建立一個連接池，然后作為參數Redis，這樣就可以實現多個Redis實例共享一個連接池。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import redis
pool = redis.ConnectionPool(host='10.211.55.4', port=6379)
r = redis.Redis(connection_pool=pool)
r.set('foo', 'Bar')
print r.get('foo')
3、管道
redis-py默認在執行每次請求都會創建（連接池申請連接）和斷開（歸還連接池）一次連接操作，如果想要在一次請求中指定多個命令，則可以使用pipline實現一次請求指定多個命令，并且默認情況下一次pipline 是原子性操作。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import redis
pool = redis.ConnectionPool(host='10.211.55.4', port=6379)
r = redis.Redis(connection_pool=pool)
# pipe = r.pipeline(transaction=False)
pipe = r.pipeline(transaction=True)
r.set('name', 'alex')
r.set('role', 'sb')
pipe.execute()
4、發布訂閱
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import redisclass RedisHelper:
def __init__(self):
self.__conn = redis.Redis(host='10.211.55.4')
self.chan_sub = 'fm104.5'
self.chan_pub = 'fm104.5'
def public(self, msg):
self.__conn.publish(self.chan_pub, msg)
return True
def subscribe(self):
pub = self.__conn.pubsub()
pub.subscribe(self.chan_sub)
pub.parse_response()
return pub
訂閱者：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from monitor.RedisHelper import RedisHelper
obj = RedisHelper()
redis_sub = obj.subscribe()
while True:
msg= redis_sub.parse_response()
print msg
發布者：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from monitor.RedisHelper import RedisHelper
obj = RedisHelper()
obj.public('hello')
更多參見：https://github.com/andymccurdy/redis-py/
RabbitMQ
RabbitMQ是一個在AMQP基礎上完整的，可復用的企業消息系統。他遵循Mozilla Public License開源協議。
MQ全稱為Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通信方法。應用程序通過讀寫出入隊列的消息（針對應用程序的數據）來通信，而無需專用連接來鏈接它們。消 息傳遞指的是程序之間通過在消息中發送數據進行通信，而不是通過直接調用彼此來通信，直接調用通常是用于諸如遠程過程調用的技術。排隊指的是應用程序通過 隊列來通信。隊列的使用除去了接收和發送應用程序同時執行的要求。
RabbitMQ安裝
安裝配置epel源
$ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm
安裝erlang
$ yum -y install erlang
安裝RabbitMQ
$ yum -y install rabbitmq-server
注意：service rabbitmq-server start/stop
安裝API
pip install pika
or
easy_install pika
or
源碼
https://pypi.python.org/pypi/pika
使用API操作RabbitMQ
基于Queue實現生產者消費者模型
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import Queue
import threading
message = Queue.Queue(10)
def producer(i):
while True:
message.put(i)
def consumer(i):
while True:
msg = message.get()
for i in range(12):
t = threading.Thread(target=producer, args=(i,))
t.start()
for i in range(10):
t = threading.Thread(target=consumer, args=(i,))
t.start()
對于RabbitMQ來說，生產和消費不再針對內存里的一個Queue對象，而是某臺服務器上的RabbitMQ Server實現的消息隊列。
#!/usr/bin/env python
import pika
# ######################### 生產者 #########################
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
channel.basic_publish(exchange='',
routing_key='hello',
body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()
#!/usr/bin/env python
import pika
# ########################## 消費者 ##########################
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
channel.basic_consume(callback,
queue='hello',
no_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
1、acknowledgment 消息不丟失
no-ack ＝ False，如果生產者遇到情況(its channel is closed, connection is closed, or TCP connection is lost)掛掉了，那么，RabbitMQ會重新將該任務添加到隊列中。
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='10.211.55.4'))
channel = connection.channel()
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
import time
time.sleep(10)
print 'ok'
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_consume(callback,
queue='hello',
no_ack=False)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
消費者
2、durable 消息不丟失
#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()
# make message persistent
channel.queue_declare(queue='hello', durable=True)
channel.basic_publish(exchange='',
routing_key='hello',
body='Hello World!',
properties=pika.BasicProperties(
delivery_mode=2, # make message persistent
))
print(" [x] Sent 'Hello World!'")
connection.close()
生產者
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()
# make message persistent
channel.queue_declare(queue='hello', durable=True)
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
import time
time.sleep(10)
print 'ok'
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_consume(callback,
queue='hello',
no_ack=False)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
消費者
3、消息獲取順序
默認消息隊列里的數據是按照順序被消費者拿走，例如：消費者1 去隊列中獲取 奇數 序列的任務，消費者1去隊列中獲取 偶數 序列的任務。
channel.basic_qos(prefetch_count=1) 表示誰來誰取，不再按照奇偶數排列
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()
# make message persistent
channel.queue_declare(queue='hello')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
import time
time.sleep(10)
print 'ok'
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
queue='hello',
no_ack=False)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
消費者
4、發布訂閱
發布訂閱和簡單的消息隊列區別在于，發布訂閱會將消息發送給所有的訂閱者，而消息隊列中的數據被消費一次便消失。所以，RabbitMQ實現發布和訂閱時，會為每一個訂閱者創建一個隊列，而發布者發布消息時，會將消息放置在所有相關隊列中。
exchange type = fanout
#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
type='fanout')
message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs',
routing_key='',
body=message)
print(" [x] Sent %r" % message)
connection.close()
發布者
#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
type='fanout')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
channel.queue_bind(exchange='logs',
queue=queue_name)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r" % body)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
訂閱者
5、關鍵字發送
exchange type = direct
之前事例，發送消息時明確指定某個隊列并向其中發送消息，RabbitMQ還支持根據關鍵字發送，即：隊列綁定關鍵字，發送者將數據根據關鍵字發送到消息exchange，exchange根據 關鍵字 判定應該將數據發送至指定隊列。
#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = sys.argv[1:]
if not severities:
sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
sys.exit(1)
for severity in severities:
channel.queue_bind(exchange='direct_logs',
queue=queue_name,
routing_key=severity)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
消費者
#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',
type='direct')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()
生產者
6、模糊匹配
exchange type = topic
在topic類型下，可以讓隊列綁定幾個模糊的關鍵字，之后發送者將數據發送到exchange，exchange將傳入”路由值“和 ”關鍵字“進行匹配，匹配成功，則將數據發送到指定隊列。
# 表示可以匹配 0 個 或 多個 單詞
* 表示只能匹配 一個 單詞
發送者路由值 隊列中
old.eddy.python old.* -- 不匹配
old.eddy.python old.# -- 匹配
#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
type='topic')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
binding_keys = sys.argv[1:]
if not binding_keys:
sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
sys.exit(1)
for binding_key in binding_keys:
channel.queue_bind(exchange='topic_logs',
queue=queue_name,
routing_key=binding_key)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
消費者
#!/usr/bin/env python
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',
type='topic')
routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
routing_key=routing_key,
body=message)
print(" [x] Sent %r:%r" % (routing_key, message))
connection.close()
生產者
SQLAlchemy
SQLAlchemy是Python編程語言下的一款ORM框架，該框架建立在數據庫API之上，使用關系對象映射進行數據庫操作，簡言之便是：將對象轉換成SQL，然后使用數據API執行SQL并獲取執行結果。
Dialect用于和數據API進行交流，根據配置文件的不同調用不同的數據庫API，從而實現對數據庫的操作，如：
MySQL-Python
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
MySQL-Connector
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
更多詳見：
步驟一：
使用 Engine/ConnectionPooling/Dialect 進行數據庫操作，Engine使用ConnectionPooling連接數據庫，然后再通過Dialect執行SQL語句。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine
engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
engine.execute(
"INSERT INTO ts_test (a, b) VALUES ('2', 'v1')"
)
engine.execute(
"INSERT INTO ts_test (a, b) VALUES (%s, %s)",
((555, "v1"),(666, "v1"),)
)
engine.execute(
"INSERT INTO ts_test (a, b) VALUES (%(id)s, %(name)s)",
id=999, name="v1"
)
result = engine.execute('select * from ts_test')
result.fetchall()
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine
engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
# 事務操作
with engine.begin() as conn:
conn.execute("insert into table (x, y, z) values (1, 2, 3)")
conn.execute("my_special_procedure(5)")
conn = engine.connect()
# 事務操作
with conn.begin():
conn.execute("some statement", {'x':5, 'y':10})
事務操作
注：查看數據庫連接：show status like 'Threads%';
步驟二：
使用 Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 進行數據庫操作。Engine使用Schema Type創建一個特定的結構對象，之后通過SQL Expression Language將該對象轉換成SQL語句，然后通過 ConnectionPooling 連接數據庫，再然后通過 Dialect 執行SQL，并獲取結果。
!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
metadata = MetaData()
user = Table('user', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(20)),
)
color = Table('color', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(20)),
)
engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
metadata.create_all(engine)
# metadata.clear()
# metadata.remove()
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
metadata = MetaData()
user = Table('user', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(20)),
)
color = Table('color', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(20)),
)
engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
conn = engine.connect()
# 創建SQL語句，INSERT INTO "user" (id, name) VALUES (:id, :name)
conn.execute(user.insert(),{'id':7,'name':'seven'})
conn.close()
# sql = user.insert().values(id=123, name='eddy')
# conn.execute(sql)
# conn.close()
# sql = user.delete().where(user.c.id > 1)
# sql = user.update().values(fullname=user.c.name)
# sql = user.update().where(user.c.name == 'jack').values(name='ed')
# sql = select([user, ])
# sql = select([user.c.id, ])
# sql = select([user.c.name, color.c.name]).where(user.c.id==color.c.id)
# sql = select([user.c.name]).order_by(user.c.name)
# sql = select([user]).group_by(user.c.name)
# result = conn.execute(sql)
# print result.fetchall()
# conn.close()
增刪改查
更多內容詳見：
http://www.jianshu.com/p/e6bba189fcbd
http://docs.sqlalchemy.org/en/latest/core/expression_api.html
注：SQLAlchemy無法修改表結構，如果需要可以使用SQLAlchemy開發者開源的另外一個軟件Alembic來完成。
步驟三：
使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有組件對數據進行操作。根據類創建對象，對象轉換成SQL，執行SQL。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(50))
# 尋找Base的所有子類，按照子類的結構在數據庫中生成對應的數據表信息
# Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# ########## 增 ##########
# u = User(id=2, name='sb')
# session.add(u)
# session.add_all([
# User(id=3, name='sb'),
# User(id=4, name='sb')
# ])
# session.commit()
# ########## 刪除 ##########
# session.query(User).filter(User.id > 2).delete()
# session.commit()
# ########## 修改 ##########
# session.query(User).filter(User.id > 2).update({'cluster_id' : 0})
# session.commit()
# ########## 查 ##########
# ret = session.query(User).filter_by(name='sb').first()
# ret = session.query(User).filter_by(name='sb').all()
# print ret
# ret = session.query(User).filter(User.name.in_(['sb','bb'])).all()
# print ret
# ret = session.query(User.name.label('name_label')).all()
# print ret,type(ret)
# ret = session.query(User).order_by(User.id).all()
# print ret
# ret = session.query(User).order_by(User.id)[1:3]
# print ret
# session.commit()
本文由用戶 Maryellen94 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。
轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。
本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！