Add `parse_maddr()` + `str` arm in `wrap_address()`
Inverse of `mk_maddr()`: parse a multiaddr string like `/ip4/127.0.0.1/tcp/1234` back into a tractor `Address`. Deats, - add `_maddr_to_tpt_proto` reverse mapping dict - add `parse_maddr()` fn dispatching on protocol combo: `[ip4|ip6, tcp]` -> `TCPAddress`, `[unix]` -> `UDSAddress` - strip leading `/` the multiaddr lib prepends to unix protocol values for correct round-trip - add `str` match case in `wrap_address()` for `/`-prefixed multiaddr strings, broaden type hint to `UnwrappedAddress|str` (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-codesubint_spawner_backend
parent
1f1e09a786
commit
90ba0e3658
|
|
@ -206,7 +206,7 @@ def mk_uuid() -> str:
|
||||||
|
|
||||||
|
|
||||||
def wrap_address(
|
def wrap_address(
|
||||||
addr: UnwrappedAddress
|
addr: UnwrappedAddress|str,
|
||||||
) -> Address:
|
) -> Address:
|
||||||
'''
|
'''
|
||||||
Wrap an `UnwrappedAddress` as an `Address`-type based
|
Wrap an `UnwrappedAddress` as an `Address`-type based
|
||||||
|
|
@ -257,6 +257,14 @@ def wrap_address(
|
||||||
cls: Type[Address] = get_address_cls(_def_tpt_proto)
|
cls: Type[Address] = get_address_cls(_def_tpt_proto)
|
||||||
addr: UnwrappedAddress = cls.get_root().unwrap()
|
addr: UnwrappedAddress = cls.get_root().unwrap()
|
||||||
|
|
||||||
|
# multiaddr-format string, e.g.
|
||||||
|
# '/ip4/127.0.0.1/tcp/1616'
|
||||||
|
case str() if addr.startswith('/'):
|
||||||
|
from tractor.discovery._multiaddr import (
|
||||||
|
parse_maddr,
|
||||||
|
)
|
||||||
|
return parse_maddr(addr)
|
||||||
|
|
||||||
case _:
|
case _:
|
||||||
# import pdbp; pdbp.set_trace()
|
# import pdbp; pdbp.set_trace()
|
||||||
# from tractor.devx import mk_pdb
|
# from tractor.devx import mk_pdb
|
||||||
|
|
|
||||||
|
|
@ -40,6 +40,12 @@ _tpt_proto_to_maddr: dict[str, str] = {
|
||||||
'uds': 'unix',
|
'uds': 'unix',
|
||||||
}
|
}
|
||||||
|
|
||||||
|
# reverse mapping: multiaddr protocol name -> tractor proto_key
|
||||||
|
_maddr_to_tpt_proto: dict[str, str] = {
|
||||||
|
v: k for k, v in _tpt_proto_to_maddr.items()
|
||||||
|
}
|
||||||
|
# {'tcp': 'tcp', 'unix': 'uds'}
|
||||||
|
|
||||||
|
|
||||||
def mk_maddr(
|
def mk_maddr(
|
||||||
addr: 'Address',
|
addr: 'Address',
|
||||||
|
|
@ -75,3 +81,46 @@ def mk_maddr(
|
||||||
return Multiaddr(
|
return Multiaddr(
|
||||||
f'/{maddr_proto}/{filepath}'
|
f'/{maddr_proto}/{filepath}'
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def parse_maddr(
|
||||||
|
maddr_str: str,
|
||||||
|
) -> 'Address':
|
||||||
|
'''
|
||||||
|
Parse a multiaddr string into a tractor `Address`.
|
||||||
|
|
||||||
|
Inverse of `mk_maddr()`.
|
||||||
|
|
||||||
|
'''
|
||||||
|
# lazy imports to avoid circular deps
|
||||||
|
from tractor.ipc._tcp import TCPAddress
|
||||||
|
from tractor.ipc._uds import UDSAddress
|
||||||
|
|
||||||
|
maddr = Multiaddr(maddr_str)
|
||||||
|
proto_names: list[str] = [
|
||||||
|
p.name for p in maddr.protocols()
|
||||||
|
]
|
||||||
|
|
||||||
|
match proto_names:
|
||||||
|
case [('ip4' | 'ip6') as net_proto, 'tcp']:
|
||||||
|
host: str = maddr.value_for_protocol(net_proto)
|
||||||
|
port: int = int(maddr.value_for_protocol('tcp'))
|
||||||
|
return TCPAddress(host, port)
|
||||||
|
|
||||||
|
case ['unix']:
|
||||||
|
# NOTE, the multiaddr lib prepends a `/` to the
|
||||||
|
# unix protocol value; strip it to recover the
|
||||||
|
# original relative path.
|
||||||
|
raw: str = maddr.value_for_protocol('unix')
|
||||||
|
sockpath = Path(raw.lstrip('/'))
|
||||||
|
return UDSAddress(
|
||||||
|
filedir=str(sockpath.parent),
|
||||||
|
filename=str(sockpath.name),
|
||||||
|
)
|
||||||
|
|
||||||
|
case _:
|
||||||
|
raise ValueError(
|
||||||
|
f'Unsupported multiaddr protocol combo: '
|
||||||
|
f'{proto_names!r}\n'
|
||||||
|
f'from maddr: {maddr_str!r}\n'
|
||||||
|
)
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue