MCPcopy Create free account
hub / github.com/bslatkin/effectivepython / AsyncServerSession

Class AsyncServerSession

example_code/item_076.py:305–362  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

303
304print("Example 17")
305class AsyncServerSession(AsyncConnection): # Changed
306 def __init__(self, *args):
307 super().__init__(*args)
308 self.clear_state()
309
310
311 print("Example 18")
312 async def loop(self): # Changed
313 while command := await self.receive(): # Changed
314 match command.split(" "):
315 case "PARAMS", lower, upper:
316 self.set_params(lower, upper)
317 case ["NUMBER"]:
318 await self.send_number() # Changed
319 case "REPORT", decision:
320 self.receive_report(decision)
321 case ["CLEAR"]:
322 self.clear_state()
323 case _:
324 raise UnknownCommandError(command)
325
326
327 print("Example 19")
328 def set_params(self, lower, upper):
329 self.clear_state()
330 self.lower = int(lower)
331 self.upper = int(upper)
332
333
334 print("Example 20")
335 def next_guess(self):
336 if self.secret is not None:
337 return self.secret
338
339 while True:
340 guess = random.randint(self.lower, self.upper)
341 if guess not in self.guesses:
342 return guess
343
344 async def send_number(self): # Changed
345 guess = self.next_guess()
346 self.guesses.append(guess)
347 await self.send(format(guess)) # Changed
348
349
350 print("Example 21")
351 def receive_report(self, decision):
352 last = self.guesses[-1]
353 if decision == CORRECT:
354 self.secret = last
355
356 print(f"Server: {last} is {decision}")
357
358 def clear_state(self):
359 self.lower = None
360 self.upper = None
361 self.secret = None
362 self.guesses = []

Callers 1

handle_async_connectionFunction · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected