diff --git a/.gitignore b/.gitignore index 77d7d6c..566c692 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,5 @@ build/ *.egg-info dist testEnv -.env \ No newline at end of file +.env +venv \ No newline at end of file diff --git a/deta/__init__.py b/deta/__init__.py index 7827a7d..707ba45 100644 --- a/deta/__init__.py +++ b/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") diff --git a/deta/base.py b/deta/base.py index 4aca041..360b84b 100644 --- a/deta/base.py +++ b/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. diff --git a/deta/drive.py b/deta/drive.py index f2452e1..2fca55b 100644 --- a/deta/drive.py +++ b/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.