Avoid weird `pydantic` runtime warning

syseng_tweaks
Tyler Goodlet 2021-05-19 16:39:59 -04:00
parent 2635ade908
commit 0b36bacfb6
2 changed files with 13 additions and 5 deletions

View File

@ -227,7 +227,7 @@ async def sample_and_broadcast(
# end up triggering backpressure which which will # end up triggering backpressure which which will
# eventually block this producer end of the feed and # eventually block this producer end of the feed and
# thus other consumers still attached. # thus other consumers still attached.
subs = bus.subscribers[sym] subs = bus._subscribers[sym]
for ctx in subs: for ctx in subs:
# print(f'sub is {ctx.chan.uid}') # print(f'sub is {ctx.chan.uid}')
try: try:

View File

@ -67,11 +67,19 @@ class _FeedsBus(BaseModel):
brokername: str brokername: str
nursery: trio.Nursery nursery: trio.Nursery
feeds: Dict[str, trio.CancelScope] = {} feeds: Dict[str, trio.CancelScope] = {}
subscribers: Dict[str, List[tractor.Context]] = {}
task_lock: trio.StrictFIFOLock = trio.StrictFIFOLock() task_lock: trio.StrictFIFOLock = trio.StrictFIFOLock()
# XXX: so weird but, apparently without this being `._` private
# pydantic will complain about private `tractor.Context` instance
# vars (namely `._portal` and `._cancel_scope`) at import time.
# Reported this bug:
# https://github.com/samuelcolvin/pydantic/issues/2816
_subscribers: Dict[str, List[tractor.Context]] = {}
class Config: class Config:
arbitrary_types_allowed = True arbitrary_types_allowed = True
underscore_attrs_are_private = False
async def cancel_all(self) -> None: async def cancel_all(self) -> None:
for sym, (cs, msg, quote) in self.feeds.items(): for sym, (cs, msg, quote) in self.feeds.items():
@ -256,7 +264,7 @@ async def attach_feed_bus(
loglevel=loglevel, loglevel=loglevel,
) )
) )
bus.subscribers.setdefault(symbol, []).append(ctx) bus._subscribers.setdefault(symbol, []).append(ctx)
else: else:
sub_only = True sub_only = True
@ -269,12 +277,12 @@ async def attach_feed_bus(
await ctx.send_yield(first_quote) await ctx.send_yield(first_quote)
if sub_only: if sub_only:
bus.subscribers[symbol].append(ctx) bus._subscribers[symbol].append(ctx)
try: try:
await trio.sleep_forever() await trio.sleep_forever()
finally: finally:
bus.subscribers[symbol].remove(ctx) bus._subscribers[symbol].remove(ctx)
@dataclass @dataclass