From 2da8f13ebdb71bedf56a6cc4df8d73625c0bb863 Mon Sep 17 00:00:00 2001 From: Balakrishnan Balasubramanian Date: Fri, 30 Jun 2023 17:42:39 -0400 Subject: [PATCH] typehint changes for python 3.8 --- mail4one/config.py | 24 ++++++++++++------------ mail4one/pop3.py | 12 ++++++------ mail4one/poputils.py | 7 ++++--- mail4one/server.py | 2 +- mail4one/smtp.py | 16 ++++++++-------- 5 files changed, 31 insertions(+), 30 deletions(-) diff --git a/mail4one/config.py b/mail4one/config.py index 1f97319..27913ac 100644 --- a/mail4one/config.py +++ b/mail4one/config.py @@ -1,14 +1,14 @@ import json import re import logging -from typing import Callable, Union, Optional +from typing import Callable, Union, Optional, List, Tuple from jata import Jata, MutableDefault class Match(Jata): name: str - addrs: list[str] = MutableDefault(lambda: []) # type: ignore - addr_rexs: list[str] = MutableDefault(lambda: []) # type: ignore + addrs: List[str] = MutableDefault(lambda: []) # type: ignore + addr_rexs: List[str] = MutableDefault(lambda: []) # type: ignore DEFAULT_MATCH_ALL = "default_match_all" @@ -23,7 +23,7 @@ class Rule(Jata): class Mbox(Jata): name: str - rules: list[Rule] + rules: List[Rule] DEFAULT_NULL_MBOX = "default_null_mbox" @@ -77,18 +77,18 @@ class Config(Jata): logging: Optional[LogCfg] = None mails_path: str - matches: list[Match] - boxes: list[Mbox] - users: list[User] + matches: List[Match] + boxes: List[Mbox] + users: List[User] - servers: list[ServerCfg] + servers: List[ServerCfg] CheckerFn = Callable[[str], bool] -Checker = tuple[str, CheckerFn, bool] +Checker = Tuple[str, CheckerFn, bool] -def parse_checkers(cfg: Config) -> list[Checker]: +def parse_checkers(cfg: Config) -> List[Checker]: def make_match_fn(m: Match): if m.addrs and m.addr_rexs: raise Exception("Both addrs and addr_rexs is set") @@ -118,7 +118,7 @@ def parse_checkers(cfg: Config) -> list[Checker]: ] -def get_mboxes(addr: str, checks: list[Checker]) -> list[str]: +def get_mboxes(addr: str, checks: List[Checker]) -> List[str]: def inner(): for mbox, match_fn, stop_check in checks: if match_fn(addr): @@ -130,7 +130,7 @@ def get_mboxes(addr: str, checks: list[Checker]) -> list[str]: return list(inner()) -def gen_addr_to_mboxes(cfg: Config) -> Callable[[str], list[str]]: +def gen_addr_to_mboxes(cfg: Config) -> Callable[[str], List[str]]: checks = parse_checkers(cfg) logging.info(f"Parsed checkers from config, {len(checks)=}") return lambda addr: get_mboxes(addr, checks) diff --git a/mail4one/pop3.py b/mail4one/pop3.py index 3214ce9..8abd569 100644 --- a/mail4one/pop3.py +++ b/mail4one/pop3.py @@ -13,7 +13,7 @@ from .pwhash import parse_hash, check_pass, PWInfo from asyncio import StreamReader, StreamWriter import random -from typing import Optional +from typing import Optional, List, Tuple, Dict from .poputils import ( InvalidCommand, @@ -46,7 +46,7 @@ class State: class SharedState: - def __init__(self, mails_path: Path, users: dict[str, tuple[PWInfo, str]]): + def __init__(self, mails_path: Path, users: dict[str, Tuple[PWInfo, str]]): self.mails_path = mails_path self.users = users self.loggedin_users: set[str] = set() @@ -237,7 +237,7 @@ def trans_command_noop(_, __) -> None: write(ok("Hmm")) -async def process_transactions(mails_list: list[MailEntry]) -> set[str]: +async def process_transactions(mails_list: List[MailEntry]) -> set[str]: mails = MailList(mails_list) def reset(_, __): @@ -328,7 +328,7 @@ async def start_session() -> None: scfg().loggedin_users.remove(state().username) -def parse_users(users: list[User]) -> dict[str, tuple[PWInfo, str]]: +def parse_users(users: List[User]) -> Dict[str, Tuple[PWInfo, str]]: def inner(): for user in users: user = User(user) @@ -338,7 +338,7 @@ def parse_users(users: list[User]) -> dict[str, tuple[PWInfo, str]]: return dict(inner()) -def make_pop_server_callback(mails_path: Path, users: list[User], timeout_seconds: int): +def make_pop_server_callback(mails_path: Path, users: List[User], timeout_seconds: int): scfg = SharedState(mails_path=mails_path, users=parse_users(users)) async def session_cb(reader: StreamReader, writer: StreamWriter): @@ -362,7 +362,7 @@ async def create_pop_server( host: str, port: int, mails_path: Path, - users: list[User], + users: List[User], ssl_context: Optional[ssl.SSLContext] = None, timeout_seconds: int = 60, ) -> asyncio.Server: diff --git a/mail4one/poputils.py b/mail4one/poputils.py index c325ff1..e8f0391 100644 --- a/mail4one/poputils.py +++ b/mail4one/poputils.py @@ -2,6 +2,7 @@ import os from dataclasses import dataclass from enum import Enum, auto from pathlib import Path +from typing import List class ClientError(Exception): @@ -112,13 +113,13 @@ def files_in_path(path): return [] -def get_mails_list(dirpath: Path) -> list[MailEntry]: +def get_mails_list(dirpath: Path) -> List[MailEntry]: files = files_in_path(dirpath) entries = [MailEntry(filename, path) for filename, path in files] return entries -def set_nid(entries: list[MailEntry]): +def set_nid(entries: List[MailEntry]): entries.sort(reverse=True, key=lambda e: (e.c_time, e.uid)) for i, entry in enumerate(entries, start=1): entry.nid = i @@ -130,7 +131,7 @@ def get_mail(entry: MailEntry) -> bytes: class MailList: - def __init__(self, entries: list[MailEntry]): + def __init__(self, entries: List[MailEntry]): self.entries = entries set_nid(self.entries) self.mails_map = {str(e.nid): e for e in entries} diff --git a/mail4one/server.py b/mail4one/server.py index 05e26c9..5698053 100644 --- a/mail4one/server.py +++ b/mail4one/server.py @@ -57,7 +57,7 @@ async def a_main(cfg: config.Config) -> None: return host mbox_finder = config.gen_addr_to_mboxes(cfg) - servers: list[asyncio.Server] = [] + servers: List[asyncio.Server] = [] if not cfg.servers: logging.warning("Nothing to do!") diff --git a/mail4one/smtp.py b/mail4one/smtp.py index fad36ce..6b2422c 100644 --- a/mail4one/smtp.py +++ b/mail4one/smtp.py @@ -7,7 +7,7 @@ import uuid import shutil from functools import partial from pathlib import Path -from typing import Callable, Optional +from typing import Callable, Optional, List from . import config from email.message import Message import email.policy @@ -25,7 +25,7 @@ logger = logging.getLogger("smtp") class MyHandler(AsyncMessage): - def __init__(self, mails_path: Path, mbox_finder: Callable[[str], list[str]]): + def __init__(self, mails_path: Path, mbox_finder: Callable[[str], List[str]]): super().__init__() self.mails_path = mails_path self.mbox_finder = mbox_finder @@ -63,7 +63,7 @@ class MyHandler(AsyncMessage): def protocol_factory_starttls( - mails_path: Path, mbox_finder: Callable[[str], list[str]], context: ssl.SSLContext + mails_path: Path, mbox_finder: Callable[[str], List[str]], context: ssl.SSLContext ): logger.info("Got smtp client cb starttls") try: @@ -80,7 +80,7 @@ def protocol_factory_starttls( return smtp -def protocol_factory(mails_path: Path, mbox_finder: Callable[[str], list[str]]): +def protocol_factory(mails_path: Path, mbox_finder: Callable[[str], List[str]]): logger.info("Got smtp client cb") try: handler = MyHandler(mails_path, mbox_finder) @@ -95,9 +95,9 @@ async def create_smtp_server_starttls( host: str, port: int, mails_path: Path, - mbox_finder: Callable[[str], list[str]], + mbox_finder: Callable[[str], List[str]], ssl_context: ssl.SSLContext, -) -> asyncio.Server: + ): logging.info( f"Starting SMTP STARTTLS server {host=}, {port=}, {mails_path=!s}, {ssl_context != None=}" ) @@ -114,9 +114,9 @@ async def create_smtp_server( host: str, port: int, mails_path: Path, - mbox_finder: Callable[[str], list[str]], + mbox_finder: Callable[[str], List[str]], ssl_context: Optional[ssl.SSLContext] = None, -) -> asyncio.Server: + ): logging.info( f"Starting SMTP server {host=}, {port=}, {mails_path=!s}, {ssl_context != None=}" )