~homeworkprod/syslog2irc

c26449abbc20f835f22b2204a657cc6cce2a4c13 — Jochen Kupperschmidt 2 months ago 34b5d74
Use PEP 585 type hinting generics
M src/syslog2irc/cli.py => src/syslog2irc/cli.py +3 -2
@@ 8,14 8,15 @@ Command line argument parsing
:License: MIT, see LICENSE for details.
"""

from __future__ import annotations
from argparse import ArgumentParser, Namespace
from pathlib import Path
from typing import List, Optional
from typing import Optional

from . import VERSION


def parse_args(args: Optional[List[str]] = None) -> Namespace:
def parse_args(args: Optional[list[str]] = None) -> Namespace:
    """Parse command line arguments."""
    parser = _create_arg_parser()
    return parser.parse_args(args)

M src/syslog2irc/config.py => src/syslog2irc/config.py +7 -6
@@ 8,10 8,11 @@ Configuration loading
:License: MIT, see LICENSE for details.
"""

from __future__ import annotations
from dataclasses import dataclass
import logging
from pathlib import Path
from typing import Any, Dict, Iterator, Optional, Set
from typing import Any, Iterator, Optional

import rtoml



@@ 35,7 36,7 @@ class ConfigurationError(Exception):
class Config:
    log_level: str
    irc: IrcConfig
    routes: Set[Route]
    routes: set[Route]


def load_config(path: Path) -> Config:


@@ 49,7 50,7 @@ def load_config(path: Path) -> Config:
    return Config(log_level=log_level, irc=irc_config, routes=routes)


def _get_log_level(data: Dict[str, Any]) -> str:
def _get_log_level(data: dict[str, Any]) -> str:
    level = data.get('log_level', 'debug').upper()

    if level not in {'CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG'}:


@@ 58,7 59,7 @@ def _get_log_level(data: Dict[str, Any]) -> str:
    return level


def _get_irc_config(data: Dict[str, Any]) -> IrcConfig:
def _get_irc_config(data: dict[str, Any]) -> IrcConfig:
    data_irc = data['irc']

    server = _get_irc_server(data_irc)


@@ 107,8 108,8 @@ def _get_irc_channels(data_irc: Any) -> Iterator[IrcChannel]:


def _get_routes(
    data: Dict[str, Any], irc_channels: Set[IrcChannel]
) -> Set[Route]:
    data: dict[str, Any], irc_channels: set[IrcChannel]
) -> set[Route]:
    data_routes = data.get('routes', {})
    if not data_routes:
        logger.warning('No routes have been configured.')

M src/syslog2irc/formatting.py => src/syslog2irc/formatting.py +2 -2
@@ 8,7 8,7 @@ Message formatting
:License: MIT, see LICENSE for details.
"""

from typing import Tuple
from __future__ import annotations

from syslogmp import Message as SyslogMessage



@@ 17,7 17,7 @@ MESSAGE_TEXT_ENCODING = 'utf-8'


def format_message(
    source_address: Tuple[str, int], message: SyslogMessage
    source_address: tuple[str, int], message: SyslogMessage
) -> str:
    """Format syslog message to be displayed on IRC."""
    source_host = source_address[0]

M src/syslog2irc/irc.py => src/syslog2irc/irc.py +7 -6
@@ 8,10 8,11 @@ Internet Relay Chat
:License: MIT, see LICENSE for details.
"""

from __future__ import annotations
from dataclasses import dataclass
import logging
import ssl
from typing import List, Optional, Set, Union
from typing import Optional, Union

from irc.bot import ServerSpec, SingleServerIRCBot
from irc.connection import Factory


@@ 49,8 50,8 @@ class IrcConfig:
    server: Optional[IrcServer]
    nickname: str
    realname: str
    commands: List[str]
    channels: Set[IrcChannel]
    commands: list[str]
    channels: set[IrcChannel]


class Bot(SingleServerIRCBot):


@@ 61,8 62,8 @@ class Bot(SingleServerIRCBot):
        server: IrcServer,
        nickname: str,
        realname: str,
        commands: List[str],
        channels: Set[IrcChannel],
        commands: list[str],
        channels: set[IrcChannel],
    ) -> None:
        logger.info(
            'Connecting to IRC server %s:%d ...', server.host, server.port


@@ 146,7 147,7 @@ class Bot(SingleServerIRCBot):
class DummyBot:
    """A fake bot that writes messages to STDOUT."""

    def __init__(self, channels: Set[IrcChannel]) -> None:
    def __init__(self, channels: set[IrcChannel]) -> None:
        self.channels = channels

    def start(self) -> None:

M src/syslog2irc/main.py => src/syslog2irc/main.py +3 -2
@@ 8,6 8,7 @@ Orchestration, application entry point
:License: MIT, see LICENSE for details.
"""

from __future__ import annotations
import logging
from queue import SimpleQueue
from typing import Callable, Optional, Tuple


@@ 74,7 75,7 @@ class Processor:
        self,
        port: Port,
        *,
        source_address: Optional[Tuple[str, int]] = None,
        source_address: Optional[tuple[str, int]] = None,
        message: Optional[SyslogMessage] = None,
    ) -> None:
        """Process an incoming syslog message."""


@@ 83,7 84,7 @@ class Processor:
    def announce_message(
        self,
        port: Port,
        source_address: Tuple[str, int],
        source_address: tuple[str, int],
        message: SyslogMessage,
    ) -> None:
        """Announce message on IRC."""

M src/syslog2irc/routing.py => src/syslog2irc/routing.py +8 -7
@@ 8,10 8,11 @@ Routing of syslog messages to IRC channels by the port they arrive on.
:License: MIT, see LICENSE for details.
"""

from __future__ import annotations
from collections import defaultdict
from dataclasses import dataclass
import logging
from typing import Any, Dict, Optional, Set
from typing import Any, Optional

from .network import format_port, Port



@@ 30,12 31,12 @@ class Route:
class Router:
    """Map syslog port numbers to IRC channel names."""

    def __init__(self, routes: Set[Route]) -> None:
    def __init__(self, routes: set[Route]) -> None:
        self.ports_to_channel_names = map_ports_to_channel_names(routes)
        self.channel_names_to_ports = map_channel_names_to_ports(
            self.ports_to_channel_names
        )
        self.enabled_channels: Set[str] = set()
        self.enabled_channels: set[str] = set()

    def enable_channel(
        self, sender: Any, *, channel_name: Optional[str] = None


@@ 59,11 60,11 @@ class Router:
    def is_channel_enabled(self, channel: str) -> bool:
        return channel in self.enabled_channels

    def get_channel_names_for_port(self, port: Port) -> Set[str]:
    def get_channel_names_for_port(self, port: Port) -> set[str]:
        return self.ports_to_channel_names[port]


def map_ports_to_channel_names(routes: Set[Route]) -> Dict[Port, Set[str]]:
def map_ports_to_channel_names(routes: set[Route]) -> dict[Port, set[str]]:
    ports_to_channel_names = defaultdict(set)
    for route in routes:
        ports_to_channel_names[route.syslog_port].add(route.irc_channel_name)


@@ 71,8 72,8 @@ def map_ports_to_channel_names(routes: Set[Route]) -> Dict[Port, Set[str]]:


def map_channel_names_to_ports(
    ports_to_channel_names: Dict[Port, Set[str]]
) -> Dict[str, Set[Port]]:
    ports_to_channel_names: dict[Port, set[str]]
) -> dict[str, set[Port]]:
    channel_names_to_ports = defaultdict(set)
    for port, channel_names in ports_to_channel_names.items():
        for channel_name in channel_names:

M src/syslog2irc/syslog.py => src/syslog2irc/syslog.py +3 -2
@@ 8,6 8,7 @@ BSD syslog message reception and handling
:License: MIT, see LICENSE for details.
"""

from __future__ import annotations
from functools import partial
import logging
from socketserver import (


@@ 17,7 18,7 @@ from socketserver import (
    ThreadingUDPServer,
)
import sys
from typing import Iterable, Tuple, Union
from typing import Iterable, Union

import syslogmp
from syslogmp import Message as SyslogMessage


@@ 71,7 72,7 @@ class UDPHandler(BaseRequestHandler):


def _handle_received_message(
    client_address: Tuple[str, int], port: Port, message: SyslogMessage
    client_address: tuple[str, int], port: Port, message: SyslogMessage
) -> None:
    logger.debug(
        'Received message from %s:%d on port %s -> %s',