new module database

apis/pastebin/.gitignore

@@ -158,3 +158,6 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+# pastebin project stuff
+*.db

apis/pastebin/db.py

@@ -0,0 +1,83 @@
+import sqlite3
+from sqlite3 import Connection
+import logging
+
+import base64, hashlib
+import datetime, time
+
+logging.basicConfig(format="%(process)d-%(levelname)s-%(message)s", level=logging.DEBUG)
+
+
+class Paste:
+    def __init__(self, paste, badge: str = ""):
+        self.id = self._hash(paste)
+        self.paste = paste
+        self.badge = badge
+        self.insert_timestamp = time.mktime(datetime.datetime.now().timetuple())
+
+    def _hash(self, val):
+        encoded_val = val.encode("utf-8")
+        h = hashlib.md5(encoded_val)
+        ret = base64.urlsafe_b64encode(h.digest()[:6])
+        return bytes.decode(ret, "utf-8")
+
+
+class DB:
+    def __init__(self, db_path: str = "default.db") -> None:
+        self._database = db_path
+        self._conn = self._create_connection(db_path)
+        self._create_pastebin_table()
+
+    def __del__(self):
+        if self._conn:
+            self._conn.close()
+
+    def insert_paste(self, paste: Paste) -> str:
+        sql = """ INSERT INTO pastebin(id,paste,insert_timestamp,badge)
+                    VALUES(?,?,?,?) """
+        try:
+            self._conn.cursor()
+            self._conn.cursor().execute(
+                sql, (paste.id, paste.paste, paste.insert_timestamp, paste.badge)
+            )
+            self._conn.commit()
+        except sqlite3.Error as e:
+            logging.error(
+                "unable to insert new paste in sqlite3 database", exc_info=True
+            )
+
+        return paste.id
+
+    def _create_pastebin_table(self) -> None:
+        sql_create_pastebin_table = """ CREATE TABLE IF NOT EXISTS pastebin (
+                                            id text PRIMARY KEY,
+                                            paste BLOB,
+                                            insert_timestamp REAL,
+                                            badge TEXT
+                                        ); """
+
+        try:
+            c = self._conn.cursor()
+            c.execute(sql_create_pastebin_table)
+        except sqlite3.Error as e:
+            logging.error("Error creating sqlite3 pastebin table", exc_info=True)
+
+    def _create_connection(self, db_file) -> Connection:
+        """returns a database connection to a SQLite database"""
+        conn = None
+        try:
+            conn = sqlite3.connect(db_file)
+            print(sqlite3.version)
+        except sqlite3.Error as e:
+            logging.error("Error creating sqlite3 database connection", exc_info=True)
+
+        return conn
+
+
+# very temporary: used to test the module locally
+#
+# if __name__ == "__main__":
+#     db = DB(r"pastebin.db")
+#     p = Paste(paste="zoom zoom", badge="mon ami")
+#     r = db.insert_paste(paste=p)
+#     print(r)

apis/pastebin/main.py

@@ -2,6 +2,10 @@
 
 from fastapi import FastAPI
 
+import logging
+
+logging.basicConfig(format="%(process)d-%(levelname)s-%(message)s", level=logging.debug)
+
 app = FastAPI()
 
 

apis/pastebin/requirements.txt

@@ -1,3 +1,4 @@
 black
 fastapi
-uvicorn[standard]
+uvicorn[standard]
+db-sqlite3