mypy: Add stubs for zephyr.
Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
parent
19f5b4f6a6
commit
34012a4015
2
mypy.ini
2
mypy.ini
|
@ -1,4 +1,6 @@
|
|||
[mypy]
|
||||
mypy_path = $MYPY_CONFIG_FILE_DIR/stubs
|
||||
|
||||
check_untyped_defs = True
|
||||
disallow_any_generics = True
|
||||
strict_optional = True
|
||||
|
|
65
stubs/_zephyr.pyi
Normal file
65
stubs/_zephyr.pyi
Normal file
|
@ -0,0 +1,65 @@
|
|||
from typing import List, Optional, Sequence, Tuple, overload
|
||||
|
||||
from typing_extensions import Literal
|
||||
|
||||
class ZUid:
|
||||
address: str
|
||||
time: float
|
||||
|
||||
class ZNotice:
|
||||
kind: int
|
||||
cls: str
|
||||
instance: str
|
||||
uid: ZUid
|
||||
time: int
|
||||
port: int
|
||||
auth: bool
|
||||
recipient: Optional[str]
|
||||
sender: Optional[str]
|
||||
opcode: Optional[str]
|
||||
format: str
|
||||
other_fields: List[str]
|
||||
fields: List[str]
|
||||
_charset: Optional[str]
|
||||
def __init__(
|
||||
self,
|
||||
kind: int = ...,
|
||||
cls: str = ...,
|
||||
instance: str = ...,
|
||||
uid: ZUid = ...,
|
||||
time: int = ...,
|
||||
port: int = ...,
|
||||
auth: bool = ...,
|
||||
recipient: Optional[str] = ...,
|
||||
sender: Optional[str] = ...,
|
||||
opcode: Optional[str] = ...,
|
||||
format: str = ...,
|
||||
other_fields: List[str] = ...,
|
||||
fields: List[str] = ...,
|
||||
_charset: Optional[str] = ...,
|
||||
message: str = ...,
|
||||
): ...
|
||||
def getmessage(self) -> str: ...
|
||||
def setmessage(self, newmsg: str) -> None: ...
|
||||
message = property(getmessage, setmessage)
|
||||
@property
|
||||
def charset(self) -> Optional[str]: ...
|
||||
def send(self) -> None: ...
|
||||
|
||||
def initialize() -> None: ...
|
||||
def openPort() -> int: ...
|
||||
def getFD() -> int: ...
|
||||
def setFD(fd: int) -> None: ...
|
||||
def sub(cls: str, instance: str, recipient: str) -> None: ...
|
||||
def subAll(lst: Sequence[Tuple[str, str, str]]) -> None: ...
|
||||
def unsub(cls: str, instance: str, recipient: str) -> None: ...
|
||||
def cancelSubs() -> None: ...
|
||||
@overload
|
||||
def receive(block: Literal[True]) -> ZNotice: ...
|
||||
@overload
|
||||
def receive(block: bool = ...) -> Optional[ZNotice]: ...
|
||||
def sender() -> str: ...
|
||||
def realm() -> str: ...
|
||||
def dump_session() -> bytes: ...
|
||||
def load_session(session: bytes) -> None: ...
|
||||
def getSubscriptions() -> List[Tuple[str, str, str]]: ...
|
13
stubs/zephyr.pyi
Normal file
13
stubs/zephyr.pyi
Normal file
|
@ -0,0 +1,13 @@
|
|||
from typing import Set, Tuple
|
||||
|
||||
import _zephyr
|
||||
from _zephyr import ZNotice as ZNotice
|
||||
from _zephyr import receive as receive
|
||||
|
||||
_z = _zephyr
|
||||
__inited: bool
|
||||
|
||||
def init() -> None: ...
|
||||
|
||||
class Subscriptions(Set[Tuple[str, str, str]]):
|
||||
pass
|
|
@ -481,7 +481,8 @@ def zephyr_load_session_autoretry(session_path: str) -> None:
|
|||
backoff = zulip.RandomExponentialBackoff()
|
||||
while backoff.keep_going():
|
||||
try:
|
||||
session = open(session_path).read()
|
||||
with open(session_path, "rb") as f:
|
||||
session = f.read()
|
||||
zephyr._z.initialize()
|
||||
zephyr._z.load_session(session)
|
||||
zephyr.__inited = True
|
||||
|
@ -523,7 +524,8 @@ def zephyr_to_zulip(options: Any) -> None:
|
|||
if options.nagios_class:
|
||||
zephyr_subscribe_autoretry((options.nagios_class, "*", "*"))
|
||||
if options.use_sessions:
|
||||
open(options.session_path, "w").write(zephyr._z.dump_session())
|
||||
with open(options.session_path, "wb") as f:
|
||||
f.write(zephyr._z.dump_session())
|
||||
|
||||
if options.logs_to_resend is not None:
|
||||
with open(options.logs_to_resend) as log:
|
||||
|
|
Loading…
Reference in a new issue