typehint changes for python 3.8

This commit is contained in:
Balakrishnan Balasubramanian 2023-06-30 17:42:39 -04:00
parent 8fe42e9163
commit 2da8f13ebd
5 changed files with 31 additions and 30 deletions

View File

@ -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)

View File

@ -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:

View File

@ -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}

View File

@ -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!")

View File

@ -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=}"
)