Python

一般是这么用的： async with aiohttp.ClientSession() as ses: res = await ses.post(xxx) text = await res.text() xxx 没有问题 但是为了减少缩进，可能想这样封装一下： class HTTP: @staticmethod async def post(*args, **kwargs): async with aiohttp.ClientSession() as ses: return await ses.post(*args, **kwargs) 这是有问题的，with 上下文之后会关闭session的连接和资源，如果payload比较大，在连接关闭之后还没读完的话，可能会卡在await ses.text()那里，导致超时 所以需要在上下文关闭之前就把内容读取完毕并返回。 可以这样： class HTTP: @staticmethod async def post(*args, **kwargs): async with aiohttp.ClientSession() as ses: async with await ses.post(*args, **kwargs) as res: return res, await res.text() 或者这样： class HTTP: @classmethod async def get(cls, *args, **kargs): await cls....

import asyncio import multiprocessing q = multiprocessing.Queue(10000) for i in range(100): q.put(i) async def coro(i): print('coro... {}'.format(i)) async def device_video_main(j): loop = asyncio.get_event_loop() for i in range(5): asyncio.ensure_future(coro(j), loop=loop) # await asyncio.sleep(1) async def run_integrate(): while True: j = q.get() print('j: ', j) if True: loop = asyncio.get_event_loop() coro = device_video_main(j) loop.create_task(coro) # asyncio.ensure_future(coro, loop=loop) # await asyncio.sleep(1) # print(loop.is_running()) # async def main(): # loop = asyncio.get_event_loop() # for i in range(3): # asyncio....

time.sleep是针对整个线程，整个线程会挂起，不再执行任何操作。 asyncio.sleep是针对当前协程而言，告诉事件循环：请去执行别的操作，相当于模拟了一次网络IO，不会阻塞其他协程的执行。 import time import asyncio async def hello(): print('Hello ...') await asyncio.sleep(5) # time.sleep(5) print('... World!') async def main(): await asyncio.gather(hello(), hello()) loop = asyncio.get_event_loop() loop.run_until_complete(main()) 运行结果： Hello ... Hello ... ... World! ... World! import time import asyncio async def hello(): print('Hello ...') # await asyncio.sleep(5) time.sleep(5) print('... World!') async def main(): await asyncio.gather(hello(), hello()) loop = asyncio.get_event_loop() loop.run_until_complete(main()) 运行结果： Hello ... ... World! Hello ... ... World!

equals query.filter(User.name == 'leela') not equals: query.filter(User.name != 'leela') LIKE query.filter(User.name.like('%leela%')) IN query.filter(User.name.in_(['leela', 'akshay', 'santanu'])) # works with query objects too: query.filter(User.name.in_(session.query(User.name).filter(User.name.like('%santanu%')))) NOT IN query.filter(~User.name.in_(['lee', 'sonal', 'akshay'])) IS NULL filter(User.name == None) IS NOT NULL filter(User.name != None) AND from sqlalchemy import and_ filter(and_(User.name == 'leela', User.fullname == 'leela dharan')) #or, default without and_ method comma separated list of conditions are AND filter(User.name == 'leela', User.fullname == 'leela dharan') # or call filter()/filter_by() multiple times filter(User....

对于UTF-8编码，Excel要求BOM(字节顺序标记)写在文件的开始，否则它会假设这是ANSI编码，这个就是与locale有依赖性了。 Python2 #!python2 #coding:utf8 import csv data = [[u'American',u'美国人'], [u'Chinese',u'中国人']] with open('results.csv','wb') as f: f.write(u'\ufeff'.encode('utf8')) w = csv.writer(f) for row in data: w.writerow([item.encode('utf8') for item in row]) Python3 #!python3 #coding:utf8 import csv data = [[u'American',u'美国人'], [u'Chinese',u'中国人']] with open('results.csv','w',newline='',encoding='utf-8-sig') as f: w = csv.writer(f) w.writerows(data) unicodecsv #!python2 #coding:utf8 import unicodecsv data = [[u'American',u'美国人'], [u'Chinese',u'中国人']] with open('results.csv','wb') as f: w = unicodecsv.writer(f,encoding='utf-8-sig') w.writerows(data) 转载自简书