File size: 2,816 Bytes
408128e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | 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")
# abstract: str = Field(..., title="Abstract")
file: UploadFile = File(..., title="File")
class Config:
schema_extra = {
"example": {
"id": "666460100c23ec4225cb2bc3",
"title": "Transformer",
# "abstract": "Kiến trúc transformer làm cho kỷ nguyên AI bùng nổ",
"file": 'bert.pdf',
"user_id": "6661455703d07f73ba",
}
}
class ProjectSchema:
def __init__(
self,
id: AnyStr = None,
title: AnyStr = "",
# abstract: AnyStr = "",
file: List = [AnyStr],
user_id: AnyStr = "",
created_at=get_current_time(),
):
self.id = id
self.title = title
# self.abstract = abstract
self.file = file
self.user_id = user_id
self.created_at = created_at
def to_dict(self):
data_dict = {
"title": self.title,
# "abstract": self.abstract,
"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)})
|