1.2.0 : support for context managers, better API

Signed-off-by: Ari Archer <[email protected]>

Author: Ari Archer

awc/__init__.py

@@ -11,7 +11,7 @@
 
 from . import const, exc, util
 
-__version__: typing.Final[str] = "1.1.0"
+__version__: typing.Final[str] = "1.2.0"
 
 
 class Awc:
@@ -54,14 +54,16 @@ def api_key(self, value: typing.Optional[str]) -> None:
         raise
             awc.exc.InvalidAPIKeyError -- on invalid API key"""
 
+        self.__api_key = value
+
         if value is not None and self.get(api="amiadmin").text != "1":
+            self.__api_key = None
+
             raise exc.InvalidAPIKeyError(
                 f"{value[:5]}{'*' * (len(value) - 5)}"[:50]
                 + (" ..." if len(value) > 50 else "")
             )
 
-        self.__api_key = value
-
     @property
     def rate_limit_wait(self) -> typing.Union[int, float]:
         """rate limit wait time getter -- get the rate limit wait in seconds
@@ -126,6 +128,12 @@ def __getitem__(self, path: str) -> str:
         return str -- the URL"""
         return self.instance.join(path).url  # type: ignore
 
+    def __enter__(self) -> "Awc":
+        return self
+
+    def __exit__(self, *_: typing.Any) -> None:
+        self.end()
+
     def request(
         self,
         method: typing.Callable[..., requests.Response],
@@ -187,6 +195,12 @@ def post(self, *args: typing.Any, **kwargs: typing.Any) -> requests.Response:
         """similar to `Awc.request` but for POST requests"""
         return self.request(self.session.post, *args, **kwargs)
 
+    def end(self) -> None:
+        """end an instance ( close it )"""
+
+        self.__api_key = None
+        self.session.close()
+
     @staticmethod
     def require_key(
         f: typing.Callable[..., typing.Any]

awc/api.py

@@ -111,7 +111,9 @@ def sql(
     raise
         awc.exc.UnexpectedResponseError from requests.exceptions.InvalidJSONError --
             on invalid JSON
-    """
+        ValueError -- on invalid queries
+
+    return typing.List[typing.List[typing.Any]] -- query results"""
 
     data: typing.Dict[str, typing.Union[typing.Iterable[str], str]] = {"sql": queries}
 
@@ -155,19 +157,19 @@ def get_comment_lock(awc: Awc) -> bool:
     """gets comments lock status
 
     awc: awc.Awc -- the awc.Awc instance to work on"""
-    return awc.get(api="lock").text == "1"
+    return util.resp_to_bool(awc.get(api="lock").text)
 
 
 @Awc.require_key
 def toggle_comment_lock(awc: Awc) -> bool:
     """toggles comments lock status
 
     awc: awc.Awc -- the awc.Awc instance to work on"""
-    return awc.post(api="lock").text == "1"
+    return util.resp_to_bool(awc.post(api="lock").text)
 
 
 def amiadmin(awc: Awc) -> bool:
     """returns your admin status ( `True` if API key is correct ( you are admin ) )
 
     awc: awc.Awc -- the awc.Awc instance to work on"""
-    return awc.get(api="amiadmin").text == "1"
+    return util.resp_to_bool(awc.get(api="amiadmin").text)

awc/sql/__init__.py

@@ -67,6 +67,33 @@ def all(cls) -> pypika.queries.QueryBuilder:
         return pypika.queries.QueryBuilder -- `SELECT * FROM self.t` query"""
         return pypika.Query.from_(cls.t).select("*")  # type: ignore
 
+    @classmethod
+    def set(
+        cls,
+        where: pypika.queries.QueryBuilder,
+        what: typing.Dict[pypika.Column, typing.Any],
+    ) -> pypika.queries.QueryBuilder:
+        """SET statement
+
+        where: pypika.queries.QueryBuilder -- the condition on which to set
+        what: dict[pypika.Column, typing.Any] -- the columns and values to set
+
+        return pypika.queries.QueryBuilder -- the query"""
+        q: pypika.queries.QueryBuilder = cls.t.update()
+
+        for k, v in what.items():
+            q = q.set(k, v)  # type: ignore
+
+        return q.where(where)  # type: ignore
+
+    @classmethod
+    def purge(cls) -> pypika.queries.QueryBuilder:
+        """purge the table ( delete everything ) **BE CAREFUL**
+
+        return pypika.queries.QueryBuilder -- the query"""
+
+        return pypika.Query.from_(cls.t).delete()  # type: ignore
+
 
 class Comment(SQLTable):
     """comment / post table"""

awc/sql/helpers.py

@@ -48,6 +48,16 @@ def unwhitelist(author: str) -> typing.List[pypika.queries.QueryBuilder]:
     ]
 
 
+def ban_ip(ip: pypika.queries.QueryBuilder) -> typing.List[pypika.queries.QueryBuilder]:
+    """ban an ip
+
+    ip: pypika.queries.QueryBuilder -- the query to get the user ip
+
+    return typing.List[pypika.queries.QueryBuilder] -- the queries"""
+
+    return [Ban.add(ip)]  # type: ignore
+
+
 def ban(author: str) -> typing.List[pypika.queries.QueryBuilder]:
     """ban and unwhitelist a user
 
@@ -58,14 +68,14 @@ def ban(author: str) -> typing.List[pypika.queries.QueryBuilder]:
     author = util.truncate(author, const.MAX_AUTHOR_LEN)
 
     return [
-        Ban.add(IpWhitelist.select(IpWhitelist.author == author, "ip"))  # type: ignore
+        *ban_ip(IpWhitelist.select(IpWhitelist.author == author, "ip"))  # type: ignore
     ] + unwhitelist(author)
 
 
 def unban(ip: str) -> typing.List[pypika.queries.QueryBuilder]:
     """unban an IP
 
-    ip: str -- the SHA256 hash of the IP bein unbannned
+    ip: str -- the SHA256 hash of the IP being unbannned
 
     return typing.List[pypika.queries.QueryBuilder] -- the queries"""
     return [delete(Ban.query(Ban.ip == ip).limit(1))]  # type: ignore
@@ -82,19 +92,11 @@ def censor_comments(
 
     return typing.List[pypika.queries.QueryBuilder] -- the queries"""
     return [
-        Comment.t.update()  # type: ignore
-        .set(Comment.author, censoring)
-        .set(Comment.content, censoring)
-        .where(where)
+        Comment.set(
+            where,
+            {
+                Comment.author: censoring,
+                Comment.content: censoring,
+            },
+        )
     ]
-
-
-def set_comments_admin(
-    where: pypika.queries.QueryBuilder,
-    value: bool = True,
-) -> typing.List[pypika.queries.QueryBuilder]:
-    """set admin status on comments
-
-    where: pypika.queries.QueryBuilder -- the condition on which to set
-    value: bool = True -- the value to set the admin status to"""
-    return [Comment.t.update().set(Comment.admin, value).where(where)]  # type: ignore

awc/util.py

@@ -45,3 +45,13 @@ def truncate(content: str, length: int, do_warn: bool = True) -> str:
         warn(ContentTruncatedWarning(content, length))
 
     return content[:length]
+
+
+def resp_to_bool(resp: str) -> bool:
+    """convets a response like 0 and 1 to boolean
+
+    resp: str -- the response
+
+    return bool -- the converted boolean"""
+
+    return resp == "1"

examples/main.py

@@ -6,8 +6,8 @@
 from warnings import filterwarnings as filter_warnings
 
 import awc
-import awc.const
 import awc.api
+import awc.const
 import awc.exc
 import awc.sql  # tip : use pypika as this library is very compatible with it :)
 import awc.sql.helpers
@@ -116,6 +116,21 @@ def main() -> int:
 
     print("you are", "an" if awc.api.amiadmin(api) else "not an", "admin")
 
+    print(f"ill call you {__name__!r} now")
+    print(
+        awc.api.sql(
+            api,
+            awc.sql.sql(
+                awc.sql.IpWhitelist.set(
+                    awc.sql.IpWhitelist.author == author,  # type: ignore
+                    {awc.sql.IpWhitelist.author: __name__},
+                )
+            ),
+        )
+    )
+
+    print("whoami api returned", (author := awc.api.whoami(api)))
+
     print("imma ban you wait")
     print(awc.api.sql(api, awc.sql.multisql(awc.sql.helpers.ban(author))))
 
@@ -126,6 +141,9 @@ def main() -> int:
     print("lol okok wait, ill unban you :) ( i wont whitelist you bc i said so !! )")
     print(awc.api.sql(api, awc.sql.multisql(awc.sql.helpers.unban(author))))
 
+    # close the connection and stuff
+    api.end()  # note : you can also use a `with` context manager
+
     return 0