Update: type hints for functions

.gitignore

@@ -7,4 +7,5 @@ build/
 *.egg-info
 dist
 testEnv
-.env
+.env
+venv

deta/__init__.py

@@ -1,10 +1,11 @@
 import os
+import typing
 import urllib.error
 import urllib.request
 import json
 
-from .base import _Base
-from .drive import _Drive
+from deta.base import _Base
+from deta.drive import _Drive
 
 
 try:
@@ -39,12 +40,17 @@ def Drive(name: str):
 
 
 class Deta:
-    def __init__(self, project_key: str = None, *, project_id: str = None):
+    def __init__(
+        self,
+        project_key: typing.Optional[str] = None,
+        *,
+        project_id: typing.Optional[str] = None,
+    ):
         project_key, project_id = _get_project_key_id(project_key, project_id)
         self.project_key = project_key
         self.project_id = project_id
 
-    def Base(self, name: str, host: str = None):
+    def Base(self, name: str, host: typing.Optional[str] = None):
         return _Base(name, self.project_key, self.project_id, host)
 
     def Drive(self, name: str, host: str = None):
@@ -59,7 +65,12 @@ def send_email(self, to, subject, message, charset="UTF-8"):
         return send_email(to, subject, message, charset)
 
 
-def send_email(to, subject, message, charset="UTF-8"):
+def send_email(
+    to: typing.Union[str, typing.List[str]],
+    subject: str,
+    message: str,
+    charset: str = "UTF-8",
+):
     pid = os.getenv("AWS_LAMBDA_FUNCTION_NAME")
     url = os.getenv("DETA_MAILER_URL")
     api_key = os.getenv("DETA_PROJECT_KEY")

deta/base.py

@@ -42,7 +42,13 @@ def prepend(self, value: typing.Union[dict, list, str, int, float, bool]):
 
 
 class _Base(_Service):
-    def __init__(self, name: str, project_key: str, project_id: str, host: str = None):
+    def __init__(
+        self,
+        name: str,
+        project_key: str,
+        project_id: str,
+        host: typing.Optional[str] = None,
+    ):
         assert name, "No Base name provided"
 
         host = host or os.getenv("DETA_BASE_HOST") or "database.deta.sh"
@@ -76,7 +82,11 @@ def delete(self, key: str):
         self._request("/items/{}".format(key), "DELETE")
         return None
 
-    def insert(self, data: typing.Union[dict, list, str, int, bool], key: str = None):
+    def insert(
+        self,
+        data: typing.Union[dict, list, str, int, bool],
+        key: typing.Optional[str] = None,
+    ):
         if not isinstance(data, dict):
             data = {"value": data}
         else:
@@ -93,7 +103,11 @@ def insert(self, data: typing.Union[dict, list, str, int, bool], key: str = None
         elif code == 409:
             raise Exception("Item with key '{4}' already exists".format(key))
 
-    def put(self, data: typing.Union[dict, list, str, int, bool], key: str = None):
+    def put(
+        self,
+        data: typing.Union[dict, list, str, int, bool],
+        key: typing.Optional[str] = None,
+    ):
         """store (put) an item in the database. Overrides an item if key already exists.
         `key` could be provided as function argument or a field in the data dict.
         If `key` is not provided, the server will generate a random 12 chars key.

deta/drive.py

@@ -19,17 +19,17 @@ def __init__(self, res: BufferedIOBase):
     def closed(self):
         return self.__stream.closed
 
-    def read(self, size: int = None):
+    def read(self, size: typing.Optional[int] = None) -> bytes:
         return self.__stream.read(size)
 
-    def iter_chunks(self, chunk_size: int = 1024):
+    def iter_chunks(self, chunk_size: int = 1024) -> typing.Iterator[bytes]:
         while True:
             chunk = self.__stream.read(chunk_size)
             if not chunk:
                 break
             yield chunk
-        
-    def iter_lines(self, chunk_size: int = 1024):
+
+    def iter_lines(self, chunk_size: int = 1024) -> typing.Iterator[bytes]:
         while True:
             chunk = self.__stream.readline(chunk_size)
             if not chunk:
@@ -64,10 +64,10 @@ def __init__(
             keep_alive=False,
         )
 
-    def _quote(self, param: str):
+    def _quote(self, param: str) -> str:
         return quote_plus(param)
 
-    def get(self, name: str):
+    def get(self, name: str) -> typing.Optional[DriveStreamingBody]:
         """Get/Download a file from drive.
         `name` is the name of the file.
         Returns a DriveStreamingBody.
@@ -80,7 +80,7 @@ def get(self, name: str):
             return DriveStreamingBody(res)
         return None
 
-    def delete_many(self, names: typing.List[str]):
+    def delete_many(self, names: typing.List[str]) -> typing.Mapping:
         """Delete many files from drive in single request.
         `names` are the names of the files to be deleted.
         Returns a dict with 'deleted' and 'failed' files.
@@ -92,7 +92,7 @@ def delete_many(self, names: typing.List[str]):
         )
         return res
 
-    def delete(self, name: str):
+    def delete(self, name: str) -> str:
         """Delete a file from drive.
         `name` is the name of the file.
         Returns the name of the file deleted.
@@ -104,7 +104,12 @@ def delete(self, name: str):
             raise Exception(f"Failed to delete '{name}':{failed[name]}")
         return name
 
-    def list(self, limit: int = 1000, prefix: str = None, last: str = None):
+    def list(
+        self,
+        limit: int = 1000,
+        prefix: typing.Optional[str] = None,
+        last: typing.Optional[str] = None,
+    ) -> typing.Mapping:
         """List file names from drive.
         `limit` is the limit of number of file names to get, defaults to 1000.
         `prefix` is the prefix  of file names.
@@ -135,7 +140,7 @@ def _upload_part(
         chunk: bytes,
         upload_id: str,
         part: int,
-        content_type: str = None,
+        content_type: typing.Optional[str] = None,
     ):
         self._request(
             f"/uploads/{upload_id}/parts?name={self._quote(name)}&part={part}",
@@ -158,8 +163,8 @@ def put(
         name: str,
         data: typing.Union[str, bytes, TextIOBase, BufferedIOBase, RawIOBase] = None,
         *,
-        path: str = None,
-        content_type: str = None,
+        path: typing.Optional[str] = None,
+        content_type: typing.Optional[str] = None,
     ) -> str:
         """Put a file in drive.
         `name` is the name of the file.