feat: modelfiles backend

backend/apps/web/models/modelfiles.py

@@ -0,0 +1,122 @@
+from pydantic import BaseModel
+from peewee import *
+from playhouse.shortcuts import model_to_dict
+from typing import List, Union, Optional
+import time
+
+from utils.utils import decode_token
+from utils.misc import get_gravatar_url
+
+from apps.web.internal.db import DB
+
+import json
+
+####################
+# User DB Schema
+####################
+
+
+class Modelfile(Model):
+    tag_name = CharField(unique=True)
+    user_id = CharField()
+    modelfile = TextField()
+    timestamp = DateField()
+
+    class Meta:
+        database = DB
+
+
+class ModelfileModel(BaseModel):
+    tag_name: str
+    user_id: str
+    modelfile: str
+    timestamp: int  # timestamp in epoch
+
+
+####################
+# Forms
+####################
+
+
+class ModelfileForm(BaseModel):
+    modelfile: dict
+
+
+class ModelfileResponse(BaseModel):
+    tag_name: str
+    user_id: str
+    modelfile: dict
+    timestamp: int  # timestamp in epoch
+
+
+class ModelfilesTable:
+    def __init__(self, db):
+        self.db = db
+        self.db.create_tables([Modelfile])
+
+    def insert_new_modelfile(
+        self, user_id: str, form_data: ModelfileForm
+    ) -> Optional[ModelfileModel]:
+        if "title" in form_data.modelfile:
+            modelfile = ModelfileModel(
+                **{
+                    "user_id": user_id,
+                    "tag_name": form_data.modelfile["title"],
+                    "modelfile": json.dumps(form_data.modelfile),
+                    "timestamp": int(time.time()),
+                }
+            )
+            result = Modelfile.create(**modelfile.model_dump())
+            if result:
+                return modelfile
+            else:
+                return None
+        else:
+            return None
+
+    def get_modelfile_by_tag_name(self, tag_name: str) -> Optional[ModelfileModel]:
+        try:
+            modelfile = Modelfile.get(Modelfile.tag_name == tag_name)
+            return ModelfileModel(**model_to_dict(modelfile))
+        except:
+            return None
+
+    def get_modelfiles(self, skip: int = 0, limit: int = 50) -> List[ModelfileResponse]:
+        return [
+            ModelfileResponse(
+                **{
+                    **model_to_dict(modelfile),
+                    "modelfile": json.loads(modelfile.modelfile),
+                }
+            )
+            for modelfile in Modelfile.select()
+            # .limit(limit).offset(skip)
+        ]
+
+    def update_modelfile_by_tag_name(
+        self, tag_name: str, modelfile: dict
+    ) -> Optional[ModelfileModel]:
+        try:
+            query = Modelfile.update(
+                modelfile=json.dumps(modelfile),
+                timestamp=int(time.time()),
+            ).where(Modelfile.tag_name == tag_name)
+
+            query.execute()
+
+            modelfile = Modelfile.get(Modelfile.tag_name == tag_name)
+            return ModelfileModel(**model_to_dict(modelfile))
+        except:
+            return None
+
+    def delete_modelfile_by_tag_name(self, tag_name: str) -> bool:
+        try:
+            query = Modelfile.delete().where((Modelfile.tag_name == tag_name))
+            query.execute()  # Remove the rows, return number of rows removed.
+
+            return True
+        except:
+            return False
+
+
+Modelfiles = ModelfilesTable(DB)