Repository: Chia-Network/chia-blockchain
File: chia/server/node_discovery.py (Python)

Code before:
    target_outbound_count: int
    peers_file_path: Path
    dns_servers: list[str]
    peer_connect_interval: int
    selected_network: str
    log: Logger
    introducer_info: Optional[dict[str, Any]] = None
    default_port: Optional[int] = None
    resolver: Optional[dns.asyncresolver.Resolver] = field(default=None, init=False)
    enable_private_networks: bool = field(default=False, init=False)
    is_closed: bool = field(default=False, init=False)
    legacy_peer_db_migrated: bool = field(default=False, init=False)
    relay_queue: Optional[asyncio.Queue[tuple[TimestampedPeerInfo, int]]] = field(default=None, init=False)
    address_manager: Optional[AddressManager] = field(default=None, init=False)
    connection_time_pretest: dict[str, Any] = field(default_factory=dict, init=False)
    received_count_from_peers: dict[str, Any] = field(default_factory=dict, init=False)
    lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)
    connect_peers_task: Optional[asyncio.Task[None]] = field(default=None, init=False)
    serialize_task: Optional[asyncio.Task[None]] = field(default=None, init=False)
    cleanup_task: Optional[asyncio.Task[None]] = field(default=None, init=False)
    initial_wait: int = field(default=0, init=False)
    pending_outbound_connections: set[str] = field(default_factory=set, init=False)
    pending_tasks: set[asyncio.Task[None]] = field(default_factory=set, init=False)
    introducer_info_obj: Optional[UnresolvedPeerInfo] = field(default=None, init=False)

    def __post_init__(self) -> None:
        random.shuffle(self.dns_servers)  # Don't always start with the same DNS server

        if self.introducer_info is not None:
            self.introducer_info_obj = UnresolvedPeerInfo(self.introducer_info["host"], self.introducer_info["port"])
            self.enable_private_networks = self.introducer_info.get("enable_private_networks", False)

        try:
            self.resolver = dns.asyncresolver.Resolver()
        except Exception:
            self.resolver = None
            self.log.exception("Error initializing asyncresolver")

        if self.default_port is None and self.selected_network in NETWORK_ID_DEFAULT_PORTS:
            self.default_port = NETWORK_ID_DEFAULT_PORTS[self.selected_network]

    async def initialize_address_manager(self) -> None:
        self.address_manager = await AddressManagerStore.create_address_manager(self.peers_file_path)
        if self.enable_private_networks:
            self.address_manager.make_private_subnets_valid()
        self.server.set_received_message_callback(self.update_peer_timestamp_on_message)

    async def start_tasks(self) -> None:
        random = Random()
        self.connect_peers_task = create_referenced_task(self._connect_to_peers(random))
        self.serialize_task = create_referenced_task(self._periodically_serialize(random))

Review comment (suggestion):
i tend away from `.__post_init__()` and towards `.create()` or `.managed()` classmethods.  this both allows you to directly instantiate with complete control when you want to and also makes it clear when you are doing calculations to create the instance.  also, a classmethod can be used consistently for both sync or async implementations unlike post init.

Code after:
    target_outbound_count: int
    peers_file_path: Path
    dns_servers: list[str]
    peer_connect_interval: int
    selected_network: str
    log: Logger
    introducer_info: Optional[dict[str, Any]] = None
    default_port: Optional[int] = None
    resolver: Optional[dns.asyncresolver.Resolver] = field(default=None)
    enable_private_networks: bool = field(default=False)
    is_closed: bool = field(default=False)
    legacy_peer_db_migrated: bool = field(default=False)
    relay_queue: Optional[asyncio.Queue[tuple[TimestampedPeerInfo, int]]] = field(default=None)
    address_manager: Optional[AddressManager] = field(default=None)
    connection_time_pretest: dict[str, Any] = field(default_factory=dict)
    received_count_from_peers: dict[str, Any] = field(default_factory=dict)
    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
    connect_peers_task: Optional[asyncio.Task[None]] = field(default=None)
    serialize_task: Optional[asyncio.Task[None]] = field(default=None)
    cleanup_task: Optional[asyncio.Task[None]] = field(default=None)
    initial_wait: int = field(default=0)
    pending_outbound_connections: set[str] = field(default_factory=set)
    pending_tasks: set[asyncio.Task[None]] = field(default_factory=set)
    introducer_info_obj: Optional[UnresolvedPeerInfo] = field(default=None)

    def __post_init__(self) -> None:
        random.shuffle(self.dns_servers)  # Don't always start with the same DNS server

        if self.introducer_info is not None:
            self.introducer_info_obj = UnresolvedPeerInfo(self.introducer_info["host"], self.introducer_info["port"])
            self.enable_private_networks = self.introducer_info.get("enable_private_networks", False)

        try:
            self.resolver = dns.asyncresolver.Resolver()
        except Exception:
            self.resolver = None
            self.log.exception("Error initializing asyncresolver")

        if self.default_port is None and self.selected_network in NETWORK_ID_DEFAULT_PORTS:
            self.default_port = NETWORK_ID_DEFAULT_PORTS[self.selected_network]

    async def initialize_address_manager(self) -> None:
        self.address_manager = await AddressManagerStore.create_address_manager(self.peers_file_path)
        if self.enable_private_networks:
            self.address_manager.make_private_subnets_valid()
        self.server.set_received_message_callback(self.update_peer_timestamp_on_message)

    async def start_tasks(self) -> None:
        random = Random()
        self.connect_peers_task = create_referenced_task(self._connect_to_peers(random))
        self.serialize_task = create_referenced_task(self._periodically_serialize(random))