Sample code for Chapter 18 - "Concurrency with asyncio"

From the book "Fluent Python" by Luciano Ramalho (O'Reilly, 2015)

http://shop.oreilly.com/product/0636920032519.do

RE_WORD=re.compile('\w+')

RE_UNICODE_NAME=re.compile('^[A-Z0-9 -]+$')

RE_CODEPOINT=re.compile('U\+([0-9A-F]{4,6})')

sample_chars= [

'$',# DOLLAR SIGN

'A',# LATIN CAPITAL LETTER A

'a',# LATIN SMALL LETTER A

'\u20a0',# EURO-CURRENCY SIGN

'\u20ac',# EURO SIGN

]

CharDescription=namedtuple('CharDescription','code_str char name')

QueryResult=namedtuple('QueryResult','count items')

deftokenize(text):

formatchinRE_WORD.finditer(text):

yieldmatch.group().upper()

defquery_type(text):

text_upper=text.upper()

if'U+'intext_upper:

elifRE_UNICODE_NAME.match(text_upper):

else:

classUnicodeNameIndex:

def__init__(self,chars=None):

self.load(chars)

defload(self,chars=None):

self.index=None

ifcharsisNone:

try:

withopen(INDEX_NAME,'rb')asfp:

self.index=pickle.load(fp)

exceptOSError:

ifself.indexisNone:

self.build_index(chars)

iflen(self.index)>MINIMUM_SAVE_LEN:

try:

self.save()

exceptOSErrorasexc:

warnings.warn('Could not save {!r}: {}'

.format(INDEX_NAME,exc))

defsave(self):

withopen(INDEX_NAME,'wb')asfp:

pickle.dump(self.index,fp)

defbuild_index(self,chars=None):

ifcharsisNone:

chars= (chr(i)foriinrange(32,sys.maxunicode))

index= {}

forcharinchars:

try:

name=unicodedata.name(char)

exceptValueError:

ifname.startswith(CJK_UNI_PREFIX):

elifname.startswith(CJK_CMP_PREFIX):

forwordintokenize(name):

index.setdefault(word,set()).add(char)

self.index=index

defword_rank(self,top=None):

res= [(len(self.index[key]),key)forkeyinself.index]

res.sort(key=lambdaitem: (-item[0],item[1]))

iftopisnotNone:

res=res[:top]

defword_report(self,top=None):

forpostings,keyinself.word_rank(top):

print('{:5} {}'.format(postings,key))

deffind_chars(self,query,start=0,stop=None):

stop=sys.maxsizeifstopisNoneelsestop

result_sets= []

forwordintokenize(query):

chars=self.index.get(word)

ifcharsisNone:# shorcut: no such word

result_sets= []

result_sets.append(chars)

ifnotresult_sets:

returnQueryResult(0, ())

result=functools.reduce(set.intersection,result_sets)

result=sorted(result)# must sort to support start, stop

result_iter=itertools.islice(result,start,stop)

returnQueryResult(len(result),

(charforcharinresult_iter))

defdescribe(self,char):

code_str='U+{:04X}'.format(ord(char))

name=unicodedata.name(char)

returnCharDescription(code_str,char,name)

deffind_descriptions(self,query,start=0,stop=None):

forcharinself.find_chars(query,start,stop).items:

yieldself.describe(char)

defget_descriptions(self,chars):

forcharinchars:

yieldself.describe(char)

defdescribe_str(self,char):

return'{:7}\t{}\t{}'.format(*self.describe(char))

deffind_description_strs(self,query,start=0,stop=None):

forcharinself.find_chars(query,start,stop).items:

yieldself.describe_str(char)

defstatus(query,counter):

ifcounter==0:

elifcounter==1:

else:

msg='{} matches'.format(counter)

return'{} for {!r}'.format(msg,query)

defmain(*args):

index=UnicodeNameIndex()

query=' '.join(args)

forn,lineinenumerate(index.find_description_strs(query),1):

print(line)

print('({})'.format(index.status(query,n)))

if__name__=='__main__':

iflen(sys.argv)>1:

main(*sys.argv[1:])

else:

print('Usage: {} word1 [word2]...'.format(sys.argv[0]))

<htmllang="en">

<metacharset="utf-8">

<title>Charfinder</title>

Examples: {links}

<formaction="/">

<inputtype="search"name="query"value="{query}">

<inputtype="submit"value="find"> {message}

{result}

SAMPLE_WORDS= ('bismillah chess cat circled Malayalam digit'

' operator Braille hexagram').split()

LINKS_HTML=', '.join(LINK_TPL.format(word)forwordin

sorted(SAMPLE_WORDS,key=str.upper))

index=UnicodeNameIndex()

withopen(TEMPLATE_NAME)astpl:

template=tpl.read()

template=template.replace('{links}',LINKS_HTML)

defhome(request):# <1>

query=request.GET.get('query','').strip()# <2>

print('Query: {!r}'.format(query))# <3>

ifquery:# <4>

descriptions=list(index.find_descriptions(query))

res='\n'.join(ROW_TPL.format(**vars(descr))

fordescrindescriptions)

msg=index.status(query,len(descriptions))

else:

descriptions= []

html=template.format(query=query,result=res,# <5>

message=msg)

print('Sending {} results'.format(len(descriptions)))# <6>

returnweb.Response(content_type=CONTENT_TYPE,text=html)# <7>

definit(loop,address,port):# <1>

app=web.Application(loop=loop)# <2>

app.router.add_route('GET','/',home)# <3>

handler=app.make_handler()# <4>

server=yieldfromloop.create_server(handler,

address,port)# <5>

returnserver.sockets[0].getsockname()# <6>

defmain(address="127.0.0.1",port=8888):

port=int(port)

loop=asyncio.get_event_loop()

host=loop.run_until_complete(init(loop,address,port))# <7>

print('Serving on {}. Hit CTRL-C to stop.'.format(host))

try:

loop.run_forever()# <8>

exceptKeyboardInterrupt:# CTRL+C pressed

print('Server shutting down.')

loop.close()# <9>

if__name__=='__main__':

main(*sys.argv[1:])

index=UnicodeNameIndex()# <2>

defhandle_queries(reader,writer):# <3>

whileTrue:# <4>

writer.write(PROMPT)# can't yield from! # <5>

yieldfromwriter.drain()# must yield from! # <6>

data=yieldfromreader.readline()# <7>

try:

query=data.decode().strip()

exceptUnicodeDecodeError:# <8>

client=writer.get_extra_info('peername')# <9>

print('Received from {}: {!r}'.format(client,query))# <10>

ifquery:

iford(query[:1])<32:# <11>

lines=list(index.find_description_strs(query))# <12>

iflines:

writer.writelines(line.encode()+CRLFforlineinlines)# <13>

writer.write(index.status(query,len(lines)).encode()+CRLF)# <14>

yieldfromwriter.drain()# <15>

print('Sent {} results'.format(len(lines)))# <16>

print('Close the client socket')# <17>

writer.close()# <18>

defmain(address='127.0.0.1',port=2323):# <1>

port=int(port)

loop=asyncio.get_event_loop()

server_coro=asyncio.start_server(handle_queries,address,port,

loop=loop)# <2>

server=loop.run_until_complete(server_coro)# <3>

host=server.sockets[0].getsockname()# <4>

print('Serving on {}. Hit CTRL-C to stop.'.format(host))# <5>

try:

loop.run_forever()# <6>

exceptKeyboardInterrupt:# CTRL+C pressed

print('Server shutting down.')

server.close()# <7>

loop.run_until_complete(server.wait_closed())# <8>

loop.close()# <9>

if__name__=='__main__':

main(*sys.argv[1:])# <10>