1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-23 11:34:18 +00:00

Improve typing

This commit is contained in:
Steffo 2020-08-05 02:42:24 +02:00
parent 743afcbc75
commit d7122cbaf5
7 changed files with 40 additions and 33 deletions

View file

@ -6,8 +6,9 @@ from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.schema import Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative.api import DeclarativeMeta, AbstractConcreteBase
from sqlalchemy.ext.declarative.api import DeclarativeMeta
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.session import Session
class Alchemy:
@ -61,7 +62,7 @@ class Alchemy:
raise TypeError(f"Can't get tables with objects of type '{table.__class__.__qualname__}'")
@contextmanager
def session_cm(self):
def session_cm(self) -> Iterator[Session]:
"""Create a Session as a context manager (that can be used in ``with`` statements).
The Session will be closed safely when the context manager exits (even in case of error).
@ -86,7 +87,7 @@ class Alchemy:
session.close()
@asynccontextmanager
async def session_acm(self):
async def session_acm(self) -> AsyncIterator[Session]:
"""Create a Session as a async context manager (that can be used in ``async with`` statements).
The Session will be closed safely when the context manager exits (even in case of error).

View file

@ -1,10 +1,13 @@
import abc
import asyncio as aio
from typing import *
from .commandargs import CommandArgs
from .commanddata import CommandData
if TYPE_CHECKING:
from .configdict import ConfigDict
from ..serf import Serf
from ..alchemy import Alchemy
class Command(metaclass=abc.ABCMeta):
@ -27,20 +30,20 @@ class Command(metaclass=abc.ABCMeta):
"""The syntax of the command, to be displayed when a :py:exc:`InvalidInputError` is raised,
in the format ``(required_arg) [optional_arg]``."""
def __init__(self, serf: "Serf", config):
def __init__(self, serf: "Serf", config: "ConfigDict"):
self.serf: "Serf" = serf
self.config = config
self.config: "ConfigDict" = config
def __str__(self):
return f"[c]{self.serf.prefix}{self.name}[/c]"
@property
def alchemy(self):
def alchemy(self) -> Alchemy:
"""A shortcut for :attr:`.interface.alchemy`."""
return self.serf.alchemy
@property
def loop(self):
def loop(self) -> aio.AbstractEventLoop:
"""A shortcut for :attr:`.interface.loop`."""
return self.serf.loop

View file

@ -4,15 +4,18 @@ from starlette.responses import Response
if TYPE_CHECKING:
from .constellation import Constellation
from ..commands import ConfigDict
from ..alchemy import Alchemy
import sqlalchemy.orm.session
class Star:
"""A Star is a class representing a part of the website.
It shouldn't be used directly: please use :class:`PageStar` and :class:`ExceptionStar` instead!"""
def __init__(self, constellation: "Constellation", config):
def __init__(self, constellation: "Constellation", config: "ConfigDict"):
self.constellation: "Constellation" = constellation
self.config = config
self.config: "ConfigDict" = config
async def page(self, request: Request) -> Response:
"""The function generating the :class:`~starlette.Response` to a web :class:`~starlette.Request`.
@ -21,13 +24,13 @@ class Star:
raise NotImplementedError()
@property
def alchemy(self):
def alchemy(self) -> Alchemy:
"""A shortcut for the :class:`~royalnet.alchemy.Alchemy` of the :class:`Constellation`."""
return self.constellation.alchemy
# noinspection PyPep8Naming
@property
def Session(self):
def Session(self) -> sqlalchemy.orm.session.Session:
"""A shortcut for the :class:`~royalnet.alchemy.Alchemy` :class:`Session` of the :class:`Constellation`."""
return self.constellation.alchemy.Session

View file

@ -25,11 +25,11 @@ class DiscordSerf(Serf):
def __init__(self,
loop: aio.AbstractEventLoop,
alchemy_cfg: Dict[str, Any],
herald_cfg: Dict[str, Any],
sentry_cfg: Dict[str, Any],
packs_cfg: Dict[str, Any],
serf_cfg: Dict[str, Any],
alchemy_cfg: rc.ConfigDict,
herald_cfg: rc.ConfigDict,
sentry_cfg: rc.ConfigDict,
packs_cfg: rc.ConfigDict,
serf_cfg: rc.ConfigDict,
**_):
if discord is None:
raise ImportError("'discord' extra is not installed")
@ -96,7 +96,7 @@ class DiscordSerf(Serf):
if not text.startswith("!"):
return
# Skip bot messages
author: Union["discord.User"] = message.author
author: Union["discord.User", "discord.Member"] = message.author
if author.bot:
return
# Find and clean parameters

View file

@ -22,11 +22,11 @@ class MatrixSerf(Serf):
def __init__(self,
loop: aio.AbstractEventLoop,
alchemy_cfg: Dict[str, Any],
herald_cfg: Dict[str, Any],
sentry_cfg: Dict[str, Any],
packs_cfg: Dict[str, Any],
serf_cfg: Dict[str, Any],
alchemy_cfg: rc.ConfigDict,
herald_cfg: rc.ConfigDict,
sentry_cfg: rc.ConfigDict,
packs_cfg: rc.ConfigDict,
serf_cfg: rc.ConfigDict,
**_):
if nio is None:
raise ImportError("'matrix' extra is not installed")

View file

@ -27,9 +27,9 @@ class Serf(abc.ABC):
def __init__(self,
loop: aio.AbstractEventLoop,
alchemy_cfg: Dict[str, Any],
herald_cfg: Dict[str, Any],
packs_cfg: Dict[str, Any],
alchemy_cfg: rc.ConfigDict,
herald_cfg: rc.ConfigDict,
packs_cfg: rc.ConfigDict,
**_):
self.loop: Optional[aio.AbstractEventLoop] = loop
"""The event loop this Serf is running on."""
@ -176,7 +176,7 @@ class Serf(abc.ABC):
raise rc.ProgramError(f"Other Herald Link returned unknown response:\n"
f"[p]{response}[/p]")
def register_commands(self, commands: List[Type[rc.Command]], pack_cfg: Dict[str, Any]) -> None:
def register_commands(self, commands: List[Type[rc.Command]], pack_cfg: rc.ConfigDict) -> None:
"""Initialize and register all commands passed as argument."""
# Instantiate the Commands
for SelectedCommand in commands:
@ -205,12 +205,12 @@ class Serf(abc.ABC):
else:
log.warning(f"Ignoring (already defined): {SelectedCommand.__qualname__} -> {alias}")
def init_herald(self, herald_cfg: Dict[str, Any]):
def init_herald(self, herald_cfg: rc.ConfigDict):
"""Create a :class:`Link` and bind :class:`Event`."""
herald_cfg["name"] = self.interface_name
self.herald: rh.Link = rh.Link(rh.Config.from_config(**herald_cfg), self.network_handler)
def register_events(self, events: List[Type[rc.HeraldEvent]], pack_cfg: Dict[str, Any]):
def register_events(self, events: List[Type[rc.HeraldEvent]], pack_cfg: rc.ConfigDict):
for SelectedEvent in events:
# Initialize the event
try:

View file

@ -39,11 +39,11 @@ class TelegramSerf(Serf):
def __init__(self,
loop: aio.AbstractEventLoop,
alchemy_cfg: Dict[str, Any],
herald_cfg: Dict[str, Any],
sentry_cfg: Dict[str, Any],
packs_cfg: Dict[str, Any],
serf_cfg: Dict[str, Any],
alchemy_cfg: rc.ConfigDict,
herald_cfg: rc.ConfigDict,
sentry_cfg: rc.ConfigDict,
packs_cfg: rc.ConfigDict,
serf_cfg: rc.ConfigDict,
**_):
if telegram is None:
raise ImportError("'telegram' extra is not installed")