type annotate twisted.application.strports

debugmm · Mar 20, 2021 · 5c38f85 · 5c38f85
1 parent 347ff6c
commit 5c38f85
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 10 deletions.
diff --git a/mypy.ini b/mypy.ini
@@ -50,10 +50,6 @@ check_untyped_defs = False
 allow_untyped_defs = True
 check_untyped_defs = False
 
-[mypy-twisted.application.strports]
-allow_untyped_defs = True
-check_untyped_defs = False
-
 [mypy-twisted.application.test.test_internet]
 allow_untyped_defs = True
 check_untyped_defs = False

diff --git a/src/twisted/application/strports.py b/src/twisted/application/strports.py
@@ -8,13 +8,23 @@
 @see: L{twisted.internet.endpoints.serverFromString}
 @see: L{twisted.internet.endpoints.clientFromString}
 """
-
+from typing import cast, Optional
 
 from twisted.application.internet import StreamServerEndpointService
-from twisted.internet import endpoints
+from twisted.internet import endpoints, interfaces
+
+
+def _getReactor() -> interfaces.IReactorCore:
+    from twisted.internet import reactor
+
+    return cast(interfaces.IReactorCore, reactor)
 
 
-def service(description, factory, reactor=None):
+def service(
+    description: str,
+    factory: interfaces.IProtocolFactory,
+    reactor: Optional[interfaces.IReactorCore] = None,
+) -> StreamServerEndpointService:
     """
     Return the service corresponding to a description.
 
@@ -33,7 +43,7 @@ def service(description, factory, reactor=None):
     @see: L{twisted.internet.endpoints.serverFromString}
     """
     if reactor is None:
-        from twisted.internet import reactor
+        reactor = _getReactor()
 
     svc = StreamServerEndpointService(
         endpoints.serverFromString(reactor, description), factory
@@ -42,7 +52,9 @@ def service(description, factory, reactor=None):
     return svc
 
 
-def listen(description, factory):
+def listen(
+    description: str, factory: interfaces.IProtocolFactory
+) -> interfaces.IListeningPort:
     """
     Listen on a port corresponding to a description.
 
@@ -63,7 +75,9 @@ def listen(description, factory):
     from twisted.internet import reactor
 
     name, args, kw = endpoints._parseServer(description, factory)
-    return getattr(reactor, "listen" + name)(*args, **kw)
+    return cast(
+        interfaces.IListeningPort, getattr(reactor, "listen" + name)(*args, **kw)
+    )
 
 
 __all__ = ["service", "listen"]