From fbf7d682f92c10ad451b169a9b5f5631f5f0c5a8 Mon Sep 17 00:00:00 2001 From: Kyomotoi Date: Fri, 1 Jul 2022 00:03:48 +0800 Subject: =?UTF-8?q?=E2=9C=A8=20=E6=9B=B4=E6=96=B0=E5=8A=9F=E8=83=BD:=20?= =?UTF-8?q?=E8=87=AA=E5=AE=9A=E4=B9=89=E8=AF=8D=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ATRI/plugins/thesaurus/db.py | 47 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 ATRI/plugins/thesaurus/db.py (limited to 'ATRI/plugins/thesaurus/db.py') diff --git a/ATRI/plugins/thesaurus/db.py b/ATRI/plugins/thesaurus/db.py new file mode 100644 index 0000000..b5394fc --- /dev/null +++ b/ATRI/plugins/thesaurus/db.py @@ -0,0 +1,47 @@ +from ATRI.database import ThesaurusStoragor, ThesaurusAuditList + + +class DBForTS: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + pass + + async def add_item(self, _id: str, group_id: int): + await ThesaurusStoragor.create(_id=_id, group_id=group_id) + + async def update_item(self, _id: str, group_id: int, update_map: dict): + await ThesaurusStoragor.filter(_id=_id, group_id=group_id).update(**update_map) + + async def del_item(self, query_map: dict): + await ThesaurusStoragor.filter(**query_map).delete() + + async def get_item_list(self, query_map: dict) -> list: + return await ThesaurusStoragor.filter(**query_map) + + async def get_all_items(self) -> list: + return await ThesaurusStoragor.all() + + +class DBForTAL: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + pass + + async def add_item(self, _id: str, group_id: int): + await ThesaurusAuditList.create(_id=_id, group_id=group_id) + + async def update_item(self, _id: str, group_id: int, update_map: dict): + await ThesaurusAuditList.filter(_id=_id, group_id=group_id).update(**update_map) + + async def del_item(self, query_map: dict): + await ThesaurusAuditList.filter(**query_map).delete() + + async def get_item_list(self, query_map: dict) -> list: + return await ThesaurusAuditList.filter(**query_map) + + async def get_all_items(self) -> list: + return await ThesaurusAuditList.all() -- cgit v1.2.3