channel.queue_declare(queue='task_queue',durable=True)

message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',# make message persistent
))
print(" [x] Sent %r" % message)
connection.close()

channel.queue_declare(queue='task_queue',durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')

def callback(ch,body):
print(" [x] Received %r" % body)
time.sleep(body.count(b'.'))
print(" [x] Done")
ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,queue='task_queue')

channel.start_consuming()

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()

channel.exchange_declare(exchange='logs',type='fanout')

result = channel.queue_declare(exclusive=True) #不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
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,body):
print(" [x] %r" % body)

channel.basic_consume(callback,queue=queue_name,no_ack=True)

channel.start_consuming()

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()

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',routing_key=severity)

print(' [*] Waiting for logs. To exit press CTRL+C')

def callback(ch,body):
print(" [x] %r:%r" % (method.routing_key,body))

channel.basic_consume(callback,no_ack=True)

channel.start_consuming()

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()

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',routing_key=binding_key)

print(' [*] Waiting for logs. To exit press CTRL+C')

def callback(ch,no_ack=True)

channel.start_consuming()

channel.queue_declare(queue='rpc_queue')

def fib(n):
if n == 0:
return 0
elif n == 1:
return 1
else:
return fib(n-1) + fib(n-2)

def on_request(ch,props,body):
n = int(body)

print(" [.] fib(%s)" % n)
response = fib(n)

ch.basic_publish(exchange='',routing_key=props.reply_to,properties=pika.BasicProperties(correlation_id = 
                                                     props.correlation_id),body=str(response))
ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request,queue='rpc_queue')

print(" Awaiting RPC requests")
channel.start_consuming()

    self.channel = self.connection.channel()

    result = self.channel.queue_declare(exclusive=True)
    self.callback_queue = result.method.queue

    self.channel.basic_consume(self.on_response,no_ack=True,queue=self.callback_queue)

def on_response(self,ch,body):
    if self.corr_id == props.correlation_id:
        self.response = body

def call(self,n):
    self.response = None
    self.corr_id = str(uuid.uuid4())
    self.channel.basic_publish(exchange='',routing_key='rpc_queue',properties=pika.BasicProperties(
                                     reply_to = self.callback_queue,correlation_id = self.corr_id,),body=str(n))
    while self.response is None:
        self.connection.process_data_events()
    return int(self.response)

fibonacci_rpc = FibonacciRpcClient()

print(" Requesting fib(30)")
response = fibonacci_rpc.call(30)
print(" [.] Got %r" % response)

event_list = []

def run():
for event in event_list:
obj = event()
obj.execute()

class BaseHandler(object):
"""
用户必须继承该类，从而规范所有类的方法（类似于接口的功能）
"""
def execute(self):
raise Exception('you must overwrite execute')

最牛逼的事件驱动框架

class MyHandler(event_drive.BaseHandler):

def execute(self):
    print 'event-drive execute MyHandler'

event_drive.event_list.append(MyHandler)
event_drive.run()

def main():
factory = protocol.ServerFactory()
factory.protocol = Echo

reactor.listenTCP(1234,factory)
reactor.run()

if name == 'main':
main()

class EchoClient(protocol.Protocol):
"""Once connected,send a message,then print the result."""

def connectionMade(self):
    self.transport.write("hello alex!")

def dataReceived(self,data):
    "As soon as any data is received,write it back."
    print "Server said:",data
    self.transport.loseConnection()

def connectionLost(self,reason):
    print "connection lost"

class EchoFactory(protocol.ClientFactory):
protocol = EchoClient

def clientConnectionFailed(self,connector,reason):
    print "Connection failed - goodbye!"
    reactor.stop()

def clientConnectionLost(self,reason):
    print "Connection lost - goodbye!"
    reactor.stop()

this connects the protocol to a server running on port 8000

def main():
f = EchoFactory()
reactor.connectTCP("localhost",1234,f)
reactor.run()

this only runs if the module was not imported

if name == 'main':
main()

from twisted.internet.protocol import ServerFactory,Protocol

def parse_args():
usage = """usage: %prog [options] poetry-file

This is the Fast Poetry Server,Twisted edition.
Run it like this:

python fastpoetry.py

If you are in the base directory of the twisted-intro package,you could run it like this:

python twisted-server-1/fastpoetry.py poetry/ecstasy.txt

to serve up John Donne's Ecstasy,which I know you want to do.
"""

parser = optparse.OptionParser(usage)

help = "The port to listen on. Default to a random available port."
parser.add_option('--port',type='int',help=help)

help = "The interface to listen on. Default is localhost."
parser.add_option('--iface',help=help,default='localhost')

options,args = parser.parse_args()
print("--arg:",options,args)

if len(args) != 1:
    parser.error('Provide exactly one poetry file.')

poetry_file = args[0]

if not os.path.exists(args[0]):
    parser.error('No such file: %s' % poetry_file)

return options,poetry_file

class PoetryProtocol(Protocol):

def connectionMade(self):
    self.transport.write(self.factory.poem)
    self.transport.loseConnection()

class PoetryFactory(ServerFactory):

protocol = PoetryProtocol

def __init__(self,poem):
    self.poem = poem

def main():
options,poetry_file = parse_args()

poem = open(poetry_file).read()

factory = PoetryFactory(poem)

from twisted.internet import reactor

port = reactor.listenTCP(options.port or 9000,factory,interface=options.iface)

print 'Serving %s on %s.' % (poetry_file,port.getHost())

reactor.run()

if name == 'main':
main()

import optparse

from twisted.internet.protocol import Protocol,ClientFactory

def parse_args():
usage = """usage: %prog [options] [hostname]:port ...

This is the Get Poetry Now! client,Twisted version 3.0
Run it like this:

python get-poetry-1.py port1 port2 port3 ...
"""

parser = optparse.OptionParser(usage)

_,addresses = parser.parse_args()

if not addresses:
    print parser.format_help()
    parser.exit()

def parse_address(addr):
    if ':' not in addr:
        host = '127.0.0.1'
        port = addr
    else:
        host,port = addr.split(':',1)

    if not port.isdigit():
        parser.error('Ports must be integers.')

    return host,int(port)

return map(parse_address,addresses)

class PoetryProtocol(Protocol):

poem = ''

def dataReceived(self,data):
    self.poem += data

def connectionLost(self,reason):
    self.poemReceived(self.poem)

def poemReceived(self,poem):
    self.factory.poem_finished(poem)

class PoetryClientFactory(ClientFactory):

protocol = PoetryProtocol

def __init__(self,callback):
    self.callback = callback

def poem_finished(self,poem):
    self.callback(poem)

def get_poetry(host,port,callback):
"""
Download a poem from the given host and port and invoke

  callback(poem)

when the poem is complete.
"""
from twisted.internet import reactor
factory = PoetryClientFactory(callback)
reactor.connectTCP(host,factory)

def poetry_main():
addresses = parse_args()

from twisted.internet import reactor

poems = []

def got_poem(poem):
    poems.append(poem)
    if len(poems) == len(addresses):
        reactor.stop()

for address in addresses:
    host,port = address
    get_poetry(host,got_poem)

reactor.run()

for poem in poems:
    print poem

if name == 'main':
poetry_main()

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,)
)
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()

metadata = MetaData()

user = Table('user',metadata,Column('id',primary_key=True),Column('name',String(20)),)

color = Table('color',)
engine = create_engine("mysql+mysqldb://root@localhost:3306/test",max_overflow=5)

metadata.create_all(engine)

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='wu')

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()

engine = create_engine("mysql+mysqldb://root@localhost:3306/test",echo=False)

class Host(Base):
tablename = 'hosts'
id = Column(Integer,primary_key=True,autoincrement=True)
hostname = Column(String(64),unique=True,nullable=False)
ip_addr = Column(String(128),nullable=False)
port = Column(Integer,default=22)

Base.metadata.create_all(engine) #创建所有表结构

if name == 'main':
SessionCls = sessionmaker(bind=engine) #创建与数据库的会话session class,注意,这里返回给session的是个class,不是实例
session = SessionCls()

h1 = Host(hostname='localhost',ip_addr='127.0.0.1')

#h2 = Host(hostname='ubuntu',ip_addr='192.168.2.243',port=20000)
#h3 = Host(hostname='ubuntu2',ip_addr='192.168.2.244',port=20000)
#session.add(h3)
#session.add_all( [h1,h2])
#h2.hostname = 'ubuntu_test' #只要没提交,此时修改也没问题
#session.rollback()
#session.commit() #提交
res = session.query(Host).filter(Host.hostname.in_(['ubuntu2','localhost'])).all()
print(res)

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()

class Child(Base):
tablename = 'child'
id = Column(Integer,primary_key=True)
parent_id = Column(Integer,ForeignKey('parent.id'))

session.query(func.count(User.name),User.name).group_by(User.name).all() SELECT count(users.name) AS count_1,users.name AS users_name
FROM users GROUP BY users.name