feature: 添加了Deta Base的支持

src/db/__init__.py

@@ -20,16 +20,14 @@ def download_file(path: str):
     print('Download database file successfully.')
 
 
-if Config.DETA and Config.database == 'sqlite3':
-    if not os.path.exists('/tmp/data.sqlite'):
-        download_file('/tmp/data.sqlite')
-elif not Config.DETA and Config.database == 'sqlite3':
+if database == 'sqlite3':
     if not os.path.exists('./src/db/data.sqlite'):
         download_file('./src/db/data.sqlite')
 
-if Config.database == 'sqlite3':
+if database == 'sqlite3':
     from src.db.db import SQLite as Database
-
+elif database == 'deta':
+    from src.db.db import DetaBase as Database
 else:
     from src.db.db import MySQL as Database
 __all__ = [Database]

src/db/db.py

@@ -6,10 +6,13 @@
 # @File Name: db.py
 
 
+import os
 from src.config import Config
 
 if Config.database == 'sqlite3':
     import sqlite3 as operator
+elif Config.database == 'deta':
+    from deta import Deta
 else:
     import pymysql as operator
 
@@ -56,9 +59,6 @@ class BaseSQL:
 class SQLite(BaseSQL):
     def __init__(self):
         super().__init__()
-        if Config.DETA:
-            self.conn = operator.connect('/tmp/data.sqlite')
-        else:
         self.conn = operator.connect('./src/db/data.sqlite')
         self.cursor = self.conn.cursor()
 
@@ -79,3 +79,66 @@ class MySQL(BaseSQL):
                                      port=port,
                                      database=db)
         self.cursor = self.conn.cursor()
+
+
+class DetaBase:
+    def __init__(self):
+        self.__deta = Deta(os.getenv("PJ_DETA"))
+        self.__data = self.__deta.Base("times")
+        self.__image = self.__deta.Base("images")
+
+    async def __get(self, _id: str) -> tuple:
+        result = self.__data.get(_id)
+        if result is None:
+            await self.__put_data(_id)
+            return tuple([_id, 0])
+        await self.update(_id, result["times"])
+        return tuple([_id, result["times"]])
+
+    async def __put_data(self, _id: str) -> bool:
+        self.__data.put({"times": 0}, _id)
+        return True
+
+    async def __update_data(self, _id: str, times: int) -> bool:
+        new = {"times": times + 1}
+        self.__data.update(new, _id)
+        return True
+
+    async def __insert_data(self, _id: str) -> bool:
+        await self.__put_data(_id)
+        return True
+
+    async def __get_images(self, theme: str) -> list:
+        response = []
+        result = self.__image.get(theme)
+        try:
+            for i in result:
+                if i != "key":
+                    response.append(tuple([
+                        i,
+                        result[i]["base64"],
+                        result[i]["width"],
+                        result[i]["height"]
+                    ]))
+        except TypeError:
+            return []
+        return response
+
+    async def query(self, _id: str) -> tuple:
+        result = await self.__get(_id)
+        return result
+
+    async def insert(self, _id: str) -> bool:
+        await self.__insert_data(_id)
+        return True
+
+    async def update(self, _id: str, times: int) -> bool:
+        await self.__update_data(_id, times)
+        return True
+
+    async def query_all(self) -> list:
+        pass
+
+    async def query_image(self, theme: str) -> list:
+        result = await self.__get_images(theme)
+        return result