|
|
from pydantic import BaseModel, Field
|
|
|
from typing import Optional, AnyStr, List, Dict
|
|
|
from ..utils.utils import get_current_time
|
|
|
from ..configs.database_config import db
|
|
|
from bson import ObjectId
|
|
|
from fastapi import UploadFile, File
|
|
|
|
|
|
collection = db["project"]
|
|
|
|
|
|
|
|
|
class Project(BaseModel):
|
|
|
title: str = Field(..., title="title")
|
|
|
|
|
|
file: UploadFile = File(..., title="File")
|
|
|
|
|
|
class Config:
|
|
|
schema_extra = {
|
|
|
"example": {
|
|
|
"id": "666460100c23ec4225cb2bc3",
|
|
|
"title": "Transformer",
|
|
|
|
|
|
"file": 'bert.pdf',
|
|
|
"user_id": "6661455703d07f73ba",
|
|
|
}
|
|
|
}
|
|
|
|
|
|
|
|
|
class ProjectSchema:
|
|
|
def __init__(
|
|
|
self,
|
|
|
id: AnyStr = None,
|
|
|
title: AnyStr = "",
|
|
|
|
|
|
file: List = [AnyStr],
|
|
|
user_id: AnyStr = "",
|
|
|
created_at=get_current_time(),
|
|
|
):
|
|
|
self.id = id
|
|
|
self.title = title
|
|
|
|
|
|
self.file = file
|
|
|
self.user_id = user_id
|
|
|
self.created_at = created_at
|
|
|
|
|
|
def to_dict(self):
|
|
|
data_dict = {
|
|
|
"title": self.title,
|
|
|
|
|
|
"file": self.file,
|
|
|
"user_id": self.user_id,
|
|
|
"created_at": self.created_at,
|
|
|
}
|
|
|
if self.id is not None:
|
|
|
data_dict["_id"] = str(self.id)
|
|
|
return data_dict
|
|
|
|
|
|
@staticmethod
|
|
|
def from_dict(data: Dict):
|
|
|
return ProjectSchema(
|
|
|
id=data.get("_id"),
|
|
|
title=data.get("title"),
|
|
|
file=data.get("file"),
|
|
|
user_id=data.get("user_id"),
|
|
|
created_at=data.get("created_at"),
|
|
|
)
|
|
|
|
|
|
def create(self, user_id: str):
|
|
|
project_dict = self.to_dict()
|
|
|
project_dict["user_id"] = user_id
|
|
|
print("datao")
|
|
|
collection.insert_one(project_dict)
|
|
|
|
|
|
@staticmethod
|
|
|
def read_all_project_by_user_id(user_id: str):
|
|
|
data = collection.find({"user_id": user_id})
|
|
|
return [ProjectSchema.from_dict(d).to_dict() for d in data]
|
|
|
|
|
|
@staticmethod
|
|
|
def read_project_by_id(project_id: str, user_id: str):
|
|
|
data = collection.find_one(
|
|
|
{"_id": ObjectId(project_id), "user_id": user_id})
|
|
|
return ProjectSchema.from_dict(data).to_dict()
|
|
|
|
|
|
def update(self, project_id: str):
|
|
|
collection.update_one(
|
|
|
{"_id": ObjectId(project_id)},
|
|
|
{"$set": self.to_dict()},
|
|
|
)
|
|
|
|
|
|
@staticmethod
|
|
|
def delete(project_id: str):
|
|
|
collection.delete_one({"_id": ObjectId(project_id)})
|
|
|
|
