hoanganh.c4i/medihome-dictionary-be
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ngay 2-8-2022

Browse Source
This commit is contained in:
Hoang Anh
2022-08-02 14:25:51 +07:00
parent 9e3a18756c
commit f541dd9956
15 changed files with 963 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
app/Dockerfile Normal file → Executable file
View File

0
app/__init__.py Normal file → Executable file
View File

20
app/main.py Normal file → Executable file
View File

@@ -1,25 +1,26 @@
#===================== Importing FastAPI necessary packages =============
# ===================== Importing FastAPI necessary packages =============
from fastapi import (
FastAPI,
HTTPException,
status,
Request,
)
from fastapi.staticfiles import StaticFiles
from src.dependecies import authenticate_user
from src.routers.routers import router
from src.routers.post import post
from src.routers.history_find import history
from src.routers.post_save import post_save
import base64
import binascii
from fastapi.middleware.cors import CORSMiddleware
#------------------ FastAPI variable ----------------------------------
# ------------------ FastAPI variable ----------------------------------
app = FastAPI()
app.mount("/post", StaticFiles(directory="post"), name="post")
origins = ["*"]
@@ -33,13 +34,13 @@ app.add_middleware(
# ================ Authentication Middleware =======================
#----------- Here authentication is based on basic scheme,
#----------- another authentication, based on bearer scheme, is used throughout
#---------- the application (as decribed in FastAPI oficial documentation)
# ----------- Here authentication is based on basic scheme,
# ----------- another authentication, based on bearer scheme, is used throughout
# ---------- the application (as decribed in FastAPI oficial documentation)
@app.middleware("http")
async def authenticate(request: Request, call_next):
#-------------------- Authentication basic scheme -----------------------------
# -------------------- Authentication basic scheme -----------------------------
if "Authorization" in request.headers:
auth = request.headers["Authorization"]
try:
@@ -60,4 +61,5 @@ async def authenticate(request: Request, call_next):
# ================= Routers inclusion from src directory ===============
app.include_router(post)
app.include_router(router)
app.include_router(history)
app.include_router(history)
app.include_router(post_save)

8
app/requirements.txt Normal file → Executable file
View File

@@ -7,7 +7,7 @@ cryptography==3.4.7
dnspython==2.1.0
ecdsa==0.17.0
email-validator==1.1.3
fastapi==0.66.0
fastapi==0.78.0
h11==0.12.0
idna==3.2
motor==2.4.0
@@ -20,8 +20,10 @@ python-jose==3.3.0
python-multipart==0.0.5
rsa==4.7.2
six==1.16.0
starlette==0.14.2
starlette==0.19.1
typing-extensions==3.10.0.0
uvicorn==0.14.0
python-multipart==0.0.5
aiofiles==0.8.0
aiofiles==0.8.0
jinja2==3.1.0
requests==2.28.1

66
app/src/dependecies.py
View File

@@ -4,14 +4,14 @@ from .settings import pwd_context, db, oauth2_scheme, SECRET_KEY, ALGORITHM
from datetime import datetime, timedelta
from typing import Optional
import json
import requests
def get_password_hash(password):
return pwd_context.hash(password)
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
@@ -21,14 +21,35 @@ async def get_user(id: str):
return user
async def authenticate_user(id: str, password: str):
user = await get_user(id)
if not user:
return False
if not verify_password(password, user["hashed_pass"]):
return False
async def authenticate_user(token: str):
url = "https://sandboxapi.ebacsi.com.vn/auth/oauth/check_token"
payload = {'token': token}
headers = {
'Authorization': 'Basic RGljdGlvbmFyeU1lZGlob21lOlJ4aXR6ZnZvaWFmZmNtb2l0ZW0='
}
response = requests.request(
"POST", url, headers=headers, data=payload)
return json.loads(response.text)
return user
async def authenticate_user_oauth2(username: str, password: str):
url = "https://sandboxapi.ebacsi.com.vn/auth/oauth/token"
payload = {'username': username,
'password': password,
'grant_type': 'password'}
files = [
]
headers = {
'Authorization': 'Basic RGljdGlvbmFyeU1lZGlob21lOlJ4aXR6ZnZvaWFmZmNtb2l0ZW0=',
'Cookie': 'JSESSIONID=node0gmjiiq3ht7kv1gesg74t1pxsb20316.node0; XSRF-TOKEN=0976f6e0-814e-4be9-b6fa-b8d0c0896315'
}
response = requests.request(
"POST", url, headers=headers, data=payload, files=files)
return response.text
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
@@ -45,19 +66,28 @@ def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
detail="User does not exist",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
url = "https://sandboxapi.ebacsi.com.vn/auth/oauth/check_token"
payload = {'token': token}
headers = {
'Authorization': 'Basic RGljdGlvbmFyeU1lZGlob21lOlJ4aXR6ZnZvaWFmZmNtb2l0ZW0='
}
response = requests.request(
"POST", url, headers=headers, data=payload)
data = json.loads(response.text)
if data is None:
raise credentials_exception
return json.loads(response.text)
# token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
user = await get_user(username)
if user is None:
raise credentials_exception
return user
# # user = await get_user(username)
# if user is None:
# raise credentials_exception
# return user

62
app/src/models/history_find.py
View File

@@ -2,6 +2,11 @@ import json
from bson import ObjectId
from pydantic import BaseModel, Field
from typing import List, Optional, Union
from ..models.post import (
PostModel,
UpdatePostModel,
ShowPostModel
)
class PyObjectId(ObjectId):
@@ -19,15 +24,15 @@ class PyObjectId(ObjectId):
def __modify_schema__(cls, field_schema):
field_schema.update(type="string")
class HistoryFindModel(BaseModel):
id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
full_name: str
username: Union[str, None] = None
age: int
sick : List[str]
token: str = None
sick: List[str] = None
authorities: List[str] = None
created_at: Optional[str] = None
post_id : str
key_find : str
post_id: str = None
key_find: str = None
class Config:
orm_mode = True
@@ -37,24 +42,50 @@ class HistoryFindModel(BaseModel):
json_encoders = {ObjectId: str}
schema_extra = {
"example": {
"full_name": "John",
"username":"",
"age": 20,
"sick": ["simple mortal"],
"post_id" : "",
"token": "",
"sick": ["sick"],
"authorities": ["ROLE_USER"],
"post_id": "",
"key_find": ""
}
}
class HistoryFindByUserModel(BaseModel):
id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
user_name: Union[str, None] = None
sick: List[str] = None
authorities: List[str] = None
created_at: Optional[str] = None
post_id: str
key_find: str
class Config:
orm_mode = True
case_sensitive = True
allow_population_by_field_name = True
arbitrary_types_allowed = True
json_encoders = {ObjectId: str}
schema_extra = {
"example": {
"user_name": "",
"sick": ["sick"],
"authorities": ["ROLE_USER"],
"post_id": "",
"key_find": ""
}
}
# class UpdateHistoryFindModel(BaseModel):
# full_name: str
# age: int
# sick : List[str]
# created_at: Optional[str] = None
# post_id : str
# key_find : str
# key_find : strfalse
# class Config:
# arbitrary_types_allowed = True
@@ -62,12 +93,7 @@ class HistoryFindModel(BaseModel):
# schema_extra = {
# "example": {
# "full_name": "John",
# "age": 20,
# "sick": "simple mortal",
# "key_find": "datetime"
# }
# }
# "age": 20,false
# class ShowPostModel(BaseModel):
# id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")

48
app/src/models/models.py
View File

@@ -1,6 +1,6 @@
from bson import ObjectId
from pydantic import BaseModel, Field
from typing import Optional, Union
from typing import Optional, Union, List
class PyObjectId(ObjectId):
@@ -25,8 +25,8 @@ class UserModel(BaseModel):
first_name: str
last_name: str
avatar: str
role : str
is_active : str
role: str
is_active: bool
created_at: Optional[str] = None
last_login: str
password: str
@@ -40,9 +40,9 @@ class UserModel(BaseModel):
"username": "",
"first_name": "John",
"last_name": "Doe",
"avatar":"",
"role": "simple mortal",
"is_active": "false",
"avatar": "",
"role": "user",
"is_active": False,
"created_at": "datetime",
"last_login": "datetime",
"password": "fakehashedsecret",
@@ -66,9 +66,9 @@ class UpdateUserModel(BaseModel):
"example": {
"first_name": "John",
"last_name": "Doe",
"avatar":"",
"role": "simple mortal",
"is_active": "false",
"avatar": "",
"role": "user",
"is_active": False,
"created_at": "datetime",
"last_login": "datetime",
}
@@ -76,25 +76,11 @@ class UpdateUserModel(BaseModel):
class ShowUserModel(BaseModel):
id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
first_name: Optional[str]
last_name: Optional[str]
avatar: str
role: Optional[str]
is_active: Optional[str]
created_at: Optional[str]
last_login: Optional[str]
class Config:
arbitrary_types_allowed = True
json_encoders = {ObjectId: str}
schema_extra = {
"example": {
"first_name": "John",
"last_name": "Doe",
"avatar":"",
"role": "simple mortal",
"created_at": "datetime",
"last_login": "datetime",
}
}
aud: List[str]
user_type: str
user_name: str
scope: List[str]
user_key: str
exp: int
authorities: List[str]
client_id: str

204
app/src/models/post.py
View File

@@ -19,34 +19,45 @@ class PyObjectId(ObjectId):
def __modify_schema__(cls, field_schema):
field_schema.update(type="string")
class token(BaseModel):
token: str
class DataPost(BaseModel):
name: str
level: str
content: List[str]
content: str
level: List[str]
class DataSmallPost(BaseModel):
name: str
level: int
point: int
level: str
class Point(BaseModel):
less10 : Optional[int] = 0
form10to20: Optional[int] = 0
form20to30: Optional[int] = 0
form30to40: Optional[int] = 0
form40to50: Optional[int] = 0
form50to60: Optional[int] = 0
bigger60: Optional[int] = 0
total: Optional[int] = 0
less10: Optional[int] = None
form10to20: Optional[int] = None
form20to30: Optional[int] = None
form30to40: Optional[int] = None
form40to50: Optional[int] = None
form50to60: Optional[int] = None
bigger60: Optional[int] = None
total: Optional[int] = None
class PostModel(BaseModel):
id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
original_post: Union[str, None] = None
translation_post: Union[str, None] = None
link : Union[str, None] = None
is_active : str
link: Union[str, None] = None
is_active: bool
created_at: Optional[str] = None
specialist: str
summary: str
data : List[DataPost]
point : Point
summary: str = None
data: List[DataPost]
point: Point = None
class Config:
orm_mode = True
case_sensitive = True
@@ -55,56 +66,36 @@ class PostModel(BaseModel):
json_encoders = {ObjectId: str}
schema_extra = {
"example": {
"original_post": "Joh111",
"translation_post": "Doe11111",
"link": "simple mortal111",
"is_active": "false",
"original_post": "Joh111",
"translation_post": "Doe11111",
"link": "simple mortal111",
"is_active": False,
"created_at": "07/20/22 02: 26: 54",
"specialist": "",
"summary": "",
"data": [
{
"name": "abc1",
"level": "user",
"content": [
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh"
]
"level": ["ORG_USER", "ORG_DICTIONARY"],
"content": "aab"
},
{
"name": "abc2",
"level": "user",
"content": [
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh"
]
"level": ["ORG_USER", "ORG_DICTIONARY"],
"content": "aab"
},
{
"name": "abc",
"level": "user",
"content": [
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh"
]
"level": ["ORG_USER", "ORG_DICTIONARY"],
"content": "aab"
},
{
"name": "abc3",
"level": "user",
"content": [
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh",
"Hoàng Anh Hoàng Anh Hoàng Anh Hoàng AnhHoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh Hoàng Anh"
]
"level": ["ORG_USER"],
"content": "aab"
}
],
"point": {
"point": {
"less10": 0,
"form10to20": 0,
"form20to30": 0,
@@ -114,33 +105,65 @@ class PostModel(BaseModel):
"bigger60": 0,
"total": 0
}
}
}
}
class UpdatePostModel(BaseModel):
original_post: Union[str, None] = None
translation_post: Union[str, None] = None
link : Union[str, None] = None
is_active : str
link: Union[str, None] = None
is_active: bool
created_at: Optional[str] = None
specialist: str
summary: str
data : List[DataPost]
point : Point
summary: str = None
data: List[DataPost]
point: Point = None
class Config:
arbitrary_types_allowed = True
json_encoders = {ObjectId: str}
schema_extra = {
"example": {
"original_post": "John",
"translation_post": "Doe",
"link": "simple mortal",
"is_active": "false",
"created_at": "datetime",
"specialist": "",
"summary": "",
"data": ""
"original_post": "Joh111",
"translation_post": "Doe11111",
"link": "simple mortal111",
"is_active": False,
"created_at": "07/20/22 02: 26: 54",
"specialist": "",
"summary": "",
"data": [
{
"name": "abc1",
"level": ["ORG_USER", "ORG_DICTIONARY"],
"content": "aab"
},
{
"name": "abc2",
"level": ["ORG_USER", "ORG_DICTIONARY"],
"content": "aab"
},
{
"name": "abc",
"level": ["ORG_USER", "ORG_DICTIONARY"],
"content": "aab"
},
{
"name": "abc3",
"level": ["ORG_USER"],
"content": "aab"
}
],
"point": {
"less10": 0,
"form10to20": 0,
"form20to30": 0,
"form30to40": 0,
"form40to50": 0,
"form50to60": 0,
"bigger60": 0,
"total": 0
}
}
}
@@ -152,20 +175,53 @@ class ShowPostModel(BaseModel):
link: Optional[str]
is_active: Optional[str]
specialist: Optional[str]
summary: Optional[str]
data : List[DataSmallPost]
point : Point
summary: Optional[str] = None
data: List[DataSmallPost]
point: Point = None
class Config:
arbitrary_types_allowed = True
json_encoders = {ObjectId: str}
schema_extra = {
"example": {
"original_post": "John",
"translation_post": "Doe",
"link": "simple mortal",
"is_active": "false",
"created_at": "datetime",
"specialist": "",
"summary": "",
"original_post": "Joh111",
"translation_post": "Doe11111",
"link": "simple mortal111",
"is_active": False,
"created_at": "07/20/22 02: 26: 54",
"specialist": "",
"summary": "",
"data": [
{
"name": "abc1",
"level": ["ORG_USER", "ORG_DICTIONARY"],
"content": "aab"
},
{
"name": "abc2",
"level": ["ORG_USER", "ORG_DICTIONARY"],
"content": "aab"
},
{
"name": "abc",
"level": ["ORG_USER", "ORG_DICTIONARY"],
"content": "aab"
},
{
"name": "abc3",
"level": ["ORG_USER"],
"content": "aab"
}
],
"point": {
"less10": 0,
"form10to20": 0,
"form20to30": 0,
"form30to40": 0,
"form40to50": 0,
"form50to60": 0,
"bigger60": 0,
"total": 0
}
}
}

69
app/src/models/save_post.py Normal file
View File

@@ -0,0 +1,69 @@
import json
from bson import ObjectId
from pydantic import BaseModel, Field
from typing import List, Optional, Union
from ..models.post import (
PostModel,
UpdatePostModel,
ShowPostModel
)
class PyObjectId(ObjectId):
@classmethod
def __get_validators__(cls):
yield cls.validate
@classmethod
def validate(cls, v):
if not ObjectId.is_valid(v):
raise ValueError("Invalid objectid")
return ObjectId(v)
@classmethod
def __modify_schema__(cls, field_schema):
field_schema.update(type="string")
class SavePostModel(BaseModel):
id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
user_name: str = None
is_active: str = None
created_at: Optional[str] = None
post_id: str = None
class Config:
orm_mode = True
case_sensitive = True
allow_population_by_field_name = True
arbitrary_types_allowed = True
json_encoders = {ObjectId: str}
schema_extra = {
"example": {
"user_name": "",
"is_active": "true",
"post_id": ""
}
}
class SavePost(BaseModel):
id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
token: str = None
is_active: str = None
created_at: Optional[str] = None
post_id: str = None
class Config:
orm_mode = True
case_sensitive = True
allow_population_by_field_name = True
arbitrary_types_allowed = True
json_encoders = {ObjectId: str}
schema_extra = {
"example": {
"token": "",
"is_active": "true",
"post_id": ""
}
}

64
app/src/routers/history_find.py
View File

@@ -11,7 +11,11 @@ from src.settings import db, ACCESS_TOKEN_EXPIRE_MINUTES
from ..models.history_find import *
from typing import List
from datetime import datetime, timedelta
from ..models.post import (
PostModel,
UpdatePostModel,
ShowPostModel
)
from ..models.models import (
UserModel,
ShowUserModel,
@@ -27,31 +31,49 @@ import re
history = APIRouter()
##############################POST###############################################
@history.post("/create_history", response_description="Add new user", response_model=HistoryFindModel)
async def create_history(history: HistoryFindModel, current_user: ShowUserModel = Depends(get_current_user)):
datetime_now = datetime.now()
post.created_at = datetime_now.strftime("%m/%d/%y %H:%M:%S")
post = jsonable_encoder(post)
new_post = await db["history_find"].insert_one(post)
created = await db["history_find"].find_one({"_id": new_post.inserted_id})
return JSONResponse(status_code=status.HTTP_201_CREATED, content=created)
@history.post("/create_history", response_description="history", response_model=HistoryFindModel)
async def create_history(history: HistoryFindModel):
datetime_now = datetime.now()
post.created_at = datetime_now.strftime("%m/%d/%y %H:%M:%S")
post = jsonable_encoder(post)
new_post = await db["history_find"].insert_one(post)
created = await db["history_find"].find_one({"_id": new_post.inserted_id})
return JSONResponse(status_code=status.HTTP_201_CREATED, content=created)
@history.get(
"/list_history", response_description="List all posts", response_model=List[HistoryFindModel]
)
async def list_post(current_user: ShowUserModel = Depends(get_current_user)):
async def list_post():
history_find = await db["history_find"].find().to_list(1000)
return history_find
# @post.get(
# "/get_post_by_name", response_description="Get a single posot", response_model=PostModel
# )
# async def show_post(id: str):
# print(id)
# post = await db["posts"].find_one({"_id": id})
# print(post)
# if post is not None:
# return post
# else:
# return None
@ history.get("/list_history_by_user", response_description="Get list Posts viewed", response_model=List[HistoryFindModel])
async def get_list_post_view_by_username(username: str, current_user: ShowUserModel = Depends(get_current_user)):
history_find = await db["history_find"].find({"username": current_user["username"]}).to_list(10)
return history_find
# @ history.get("/list_post_by_user", response_description="Get list Posts viewed", response_model=List[ShowPostModel])
# async def get_list_post_view_by_username(username: str, current_user: ShowUserModel = Depends(get_current_user)):
# history_find = await db["history_find"].find({"username": username}).to_list(10)
# post_view = []
# for dt in history_find:
# print(dt["post_id"])
# post = await db["posts"].find_one({"_id": dt["post_id"]})
# post_view.append(post)
# return post_view
@history.get("/list_post_by_user", response_description="Get list Posts viewed", response_model=List[HistoryFindByUserModel])
async def get_list_post_view_by_username(current_user: ShowUserModel = Depends(get_current_user)):
history_find = await db["history_find"].find({"username": current_user["username"]}).to_list(10)
post_view = []
for dt in history_find:
print(dt["post_id"])
post = await db["posts"].find_one({"_id": dt["post_id"]})
dt["post"] = post
return history_find

336
app/src/routers/post.py
View File

@@ -1,8 +1,9 @@
from email.policy import default
from fastapi import APIRouter,Depends,status,HTTPException,UploadFile, File, Header
from fastapi import APIRouter, Depends, status, HTTPException, UploadFile, File, Header, Request
from fastapi.responses import JSONResponse, FileResponse, StreamingResponse
from fastapi.encoders import jsonable_encoder
from fastapi.security import OAuth2PasswordRequestForm
from fastapi.staticfiles import StaticFiles
import logging
from ..dependecies import (
get_current_user,
@@ -32,137 +33,238 @@ from typing import List
from datetime import datetime
import os
import re
from pathlib import Path
import codecs
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
post = APIRouter()
##############################POST###############################################
@post.post("/create_post", response_description="Add new user", response_model=PostModel)
async def create_post(post: PostModel, current_user: UserModel = Depends(get_current_user)):
@post.get("/post_html", response_class=HTMLResponse)
async def post_html(content: str, request: Request):
templates = Jinja2Templates(directory="post")
return templates.TemplateResponse(content, {"request": request})
@post.post("/create_post", response_description="Add new post", response_model=PostModel)
async def create_post(post: PostModel,
# current_user: UserModel = Depends(get_current_user)
):
try:
print(post)
# if current_user["role"] == "user":
# return JSONResponse(content="Role User không được phép tạo bài viết")
datetime_now = datetime.now()
post.created_at = datetime_now.strftime("%m/%d/%y %H:%M:%S")
post = jsonable_encoder(post)
if(post.get("point", None) == None):
post["point"] = {
"less10": 0,
"form10to20": 0,
"form20to30": 0,
"form30to40": 0,
"form40to50": 0,
"form50to60": 0,
"bigger60": 0,
"total": 0
}
new_post = await db["posts"].insert_one(post)
created_post = await db["posts"].find_one({"_id": new_post.inserted_id})
return JSONResponse(status_code=status.HTTP_201_CREATED, content=created_post)
except NameError:
return NameError
@post.get(
"/list_post", response_description="List all posts", response_model=List[ShowPostModel]
"/list_post", response_description="List all posts"
)
async def list_post(current_user: UserModel = Depends(get_current_user)):
async def list_post():
posts = await db["posts"].find().to_list(1000)
print(posts)
print(len(posts))
return posts
@post.post(
"/find_list_post", response_description="search list posts", response_model=List[ShowPostModel]
)
async def list_post(history: HistoryFindModel, current_user: UserModel = Depends(get_current_user)):
async def list_post(key_find: str, data: str = None):
point_data=["point.less10",
"point.form10to20",
"point.form20to30",
"point.form30to40",
"point.form40to50",
"point.form50to60",
]
# point_data = ["point.less10",
# "point.form10to20",
# "point.form20to30",
# "point.form30to40",
# "point.form40to50",
# "point.form50to60",
# ]
age_sort = "point.total"
history = jsonable_encoder(history)
if history.get("age", None) != None:
if history.get("age") >59:
age_sort = "point.bigger60"
else:
age_sort = point_data[history.get("age")//10]
posts = await db["posts"].find({"translation_post": { "$regex": history.get("key_find") } }).sort(age_sort, -1).to_list(100)
# history = jsonable_encoder(history)
# if history.get("age", None) != None:
# if history.get("age") > 59:
# age_sort = "point.bigger60"
# else:
# age_sort = point_data[history.get("age")//10]
posts = await db["posts"].find({"translation_post": {"$regex": key_find}}).sort(age_sort, -1).to_list(100)
print(posts)
return posts
@post.post(
"/get_post_by_name", response_description="Get a single post", response_model=PostModel
"/get_post_by_name"
# , response_description="Get a single post", response_model=PostModel
)
async def get_post_by_name(history: HistoryFindModel, current_user: UserModel = Depends(get_current_user)
):
datetime_now = datetime.now()
history.created_at = datetime_now.strftime("%m/%d/%y %H:%M:%S")
post = await db["posts"].find_one({"_id": history.post_id})
history = jsonable_encoder(history)
new_his = await db["history_find"].insert_one(history)
created = await db["history_find"].find_one({"_id": new_his.inserted_id})
print(created)
if post is not None:
return post
else:
return None
async def get_post_by_name(history: HistoryFindModel):
try:
history = jsonable_encoder(history)
token = history.get("token", None)
if token == '':
datetime_now = datetime.now()
history["created_at"] = datetime_now.strftime("%m/%d/%y %H:%M:%S")
post = await db["posts"].find_one({"_id": history["post_id"]})
print(post)
for dt in post["data"]:
if dt["level"] != ["*"]:
dt["content"] = "Bạn không có quyền xem nội dung này, vui lòng sử dụng tải khoản được cấp quyền để xem nội dung"
history = jsonable_encoder(history)
new_his = await db["history_find"].insert_one(history)
created = await db["history_find"].find_one({"_id": new_his.inserted_id})
if post is not None:
return post
else:
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={'message': "Bài viết không tồn tại"}
)
data_token = await get_current_user(history["token"])
data = data_token.get("user_name", None)
user_type = data_token.get("user_type", None)
if data == None:
return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED, content={"message": "UNAUTHORIZED"})
else:
datetime_now = datetime.now()
history["created_at"] = datetime_now.strftime("%m/%d/%y %H:%M:%S")
post = await db["posts"].find_one({"_id": history["post_id"]})
for dt in post["data"]:
if dt["level"] != ["*"]:
if not user_type in dt["level"]:
dt["content"] = "Bạn không có quyền xem nội dung này, vui lòng sử dụng tải khoản được cấp quyền để xem nội dung"
history = jsonable_encoder(history)
new_his = await db["history_find"].insert_one(history)
created = await db["history_find"].find_one({"_id": new_his.inserted_id})
if post is not None:
return post
else:
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={'message': "Bài viết không tồn tại"}
)
except Exception as e:
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={'message': str(e)}
)
@post.post(
"/get_post_edit"
)
async def get_post_edit(history: HistoryFindModel):
try:
datetime_now = datetime.now()
history = jsonable_encoder(history)
history["created_at"] = datetime_now.strftime("%m/%d/%y %H:%M:%S")
post = await db["posts"].find_one({"_id": history["post_id"]})
new_his = await db["history_find"].insert_one(history)
created = await db["history_find"].find_one({"_id": new_his.inserted_id})
if post is not None:
return post
else:
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={'message': "Bài viết không tồn tại"}
)
except Exception as e:
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={'message': str(e)}
)
@post.delete("/delete_post/{id}", response_description="Delete a post")
async def delete_user(id: str, current_user: UserModel = Depends(get_current_user)):
async def delete_user(id: str,
# current_user: UserModel = Depends(get_current_user)
):
delete_result = await db["posts"].delete_one({"_id": id})
if delete_result.deleted_count == 1:
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
raise HTTPException(status_code=404, detail=f"Post {id} not found")
@post.get("/score", response_description="score all post", response_model=List[ShowPostModel])
async def score_all_post( current_user: UserModel = Depends(get_current_user)):
async def score_all_post(
# current_user: UserModel = Depends(get_current_user)
):
posts = await db["posts"].find().to_list(1000)
for dt_post in posts:
print(dt_post)
data_old = dt_post
dt_post["point"]["less10"] = await db["history_find"].count_documents({"post_id": dt_post["_id"],
"age":{
'$gte': 0, '$lte': 9
}})
"age": {
'$gte': 0, '$lte': 9
}})
dt_post["point"]["form10to20"] = await db["history_find"].count_documents({"post_id": dt_post["_id"],
"age":{
'$gte': 10, '$lte': 19
}})
"age": {
'$gte': 10, '$lte': 19
}})
dt_post["point"]["form20to30"] = await db["history_find"].count_documents({"post_id": dt_post["_id"],
"age":{
'$gte': 20, '$lte': 29
}})
"age": {
'$gte': 20, '$lte': 29
}})
dt_post["point"]["form30to40"] = await db["history_find"].count_documents({"post_id": dt_post["_id"],
"age":{
'$gte': 30, '$lte': 39
}})
"age": {
'$gte': 30, '$lte': 39
}})
dt_post["point"]["form40to50"] = await db["history_find"].count_documents({"post_id": dt_post["_id"],
"age":{
'$gte': 40, '$lte': 49
}})
"age": {
'$gte': 40, '$lte': 49
}})
dt_post["point"]["form50to60"] = await db["history_find"].count_documents({"post_id": dt_post["_id"],
"age":{
'$gte': 50, '$lte': 59
}})
"age": {
'$gte': 50, '$lte': 59
}})
dt_post["point"]["bigger60"] = await db["history_find"].count_documents({"post_id": dt_post["_id"],
"age":{
'$gte': 60
}})
"age": {
'$gte': 60
}})
dt_post["point"]["total"] = await db["history_find"].count_documents({"post_id": dt_post["_id"]})
# await db["posts"].update_one(data_old, dt_post)
await db["posts"].update_one({"_id": dt_post["_id"]}, {"$set": {
"point":{
"less10": dt_post["point"]["less10"],
"form10to20": dt_post["point"]["form10to20"],
"form20to30": dt_post["point"]["form20to30"],
"form30to40": dt_post["point"]["form30to40"],
"form40to50": dt_post["point"]["form40to50"],
"form50to60": dt_post["point"]["form50to60"],
"bigger60": dt_post["point"]["bigger60"],
"total": dt_post["point"]["total"]
}
"point": {
"less10": dt_post["point"]["less10"],
"form10to20": dt_post["point"]["form10to20"],
"form20to30": dt_post["point"]["form20to30"],
"form30to40": dt_post["point"]["form30to40"],
"form40to50": dt_post["point"]["form40to50"],
"form50to60": dt_post["point"]["form50to60"],
"bigger60": dt_post["point"]["bigger60"],
"total": dt_post["point"]["total"]
}
}})
return posts
return posts
@post.post("/uploadfiles/")
async def create_upload_files(
files: List[UploadFile] = File(...), current_user: UserModel = Depends(get_current_user)
files: List[UploadFile] = File(...),
# current_user: UserModel = Depends(get_current_user)
):
try:
file_name =[]
file_name = []
i = 0
file_location = f"../media/"
for file in files:
@@ -172,31 +274,107 @@ async def create_upload_files(
file_name.append(current_time + str(i) + file.filename)
i = i + 1
with open(file_save, "wb+") as file_object:
file_object.write(file.file.read())
file_object.write(file.file.read())
return {"filenames": file_name}
except Exception as e:
return JSONResponse(
status_code = status.HTTP_400_BAD_REQUEST,
content = { 'message' : str(e) }
)
status_code=status.HTTP_400_BAD_REQUEST,
content={'message': str(e)}
)
@post.post("/upload_post/")
async def create_upload_post(
post: List[UploadFile] = File(...),
image: List[UploadFile] = File(...),
# current_user: UserModel = Depends(get_current_user)
):
try:
file_name = []
i = 0
now = datetime.now()
current_time = now.strftime("%H_%M_%S_%d-%m-%Y_")
folder_save = f"./post/" + current_time + str(i)
Path(folder_save).mkdir(parents=True)
for file in post:
file_save = folder_save + "/" + file.filename
file_name.append(current_time + str(i) + "/" + file.filename)
i = i + 1
with open(file_save, "wb+") as file_object:
file_object.write(file.file.read())
folder_save = folder_save + "/images"
Path(folder_save).mkdir(parents=True, exist_ok=True)
for file in image:
file_save = folder_save + "/" + file.filename
i = i + 1
with open(file_save, "wb+") as file_object:
file_object.write(file.file.read())
return {"filenames": file_name}
except Exception as e:
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={'message': str(e)}
)
# @post.get("/show_post")
# async def show_file(name: str):
# try:
# f = codecs.open(
# "/app/post/04_52_34_27-07-2022_0/formatted.html", "r", "utf-8")
# f.read()
# return {"data": f.read()}
# except Exception as e:
# return JSONResponse(
# status_code=status.HTTP_400_BAD_REQUEST,
# content={'message': str(e)}
# )
@post.get("/showfile/{name}")
async def show_file(name: str, current_user: UserModel = Depends(get_current_user)):
async def show_file(name: str
# , current_user: UserModel = Depends(get_current_user)
):
file_path = f"../media/" + name
if os.path.exists(file_path):
return FileResponse(file_path)
return {"erro": "File not found!"}
@post.get("/showvideo/{file_name}", response_class=FileResponse)
async def main(file_name: str, current_user: UserModel = Depends(get_current_user)):
async def main(file_name: str
# , current_user: UserModel = Depends(get_current_user)
):
file_path = f"../media/" + file_name
return file_path
@post.get("/video")
async def video_endpoint(video_name, current_user: UserModel = Depends(get_current_user)):
async def video_endpoint(video_name
# , current_user: UserModel = Depends(get_current_user)
):
def iterfile():
video_path = f"../media/" + video_name
with open(video_path, mode="rb") as file_like:
yield from file_like
return StreamingResponse(iterfile(), media_type="video/mp4")
return StreamingResponse(iterfile(), media_type="video/mp4")
@post.post("/edit_post/{id}", response_description="score all post", response_model=UpdatePostModel)
async def edit_post(id: str, post: UpdatePostModel):
try:
post = jsonable_encoder(post)
await db["posts"].update_one({"_id": id}, {"$set":
post
})
created_post = await db["posts"].find_one({"_id": id})
return JSONResponse(status_code=status.HTTP_200_OK, content=created_post)
except Exception as e:
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={'message': str(e)}
)

85
app/src/routers/post_save.py Normal file
View File

@@ -0,0 +1,85 @@
from os import access
from urllib import response
from fastapi import (
APIRouter,
Depends,
status,
HTTPException
)
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder
from fastapi.security import OAuth2PasswordRequestForm
from fastapi import File, UploadFile, FastAPI
from ..models.models import (
UserModel,
ShowUserModel,
UpdateUserModel
)
from ..models.save_post import *
from ..dependecies import (
get_current_user,
authenticate_user,
authenticate_user_oauth2,
create_access_token,
get_password_hash
)
from ..settings import db, ACCESS_TOKEN_EXPIRE_MINUTES
import json
from typing import List
from datetime import datetime, timedelta
import requests
import re
from pydantic import BaseModel, Field
post_save = APIRouter()
# ============= Creating path operations ==============
@post_save.post("/save_post", response_description="save new post", response_model=SavePostModel)
async def create_post(post_save: SavePost):
try:
url = "https://sandboxapi.ebacsi.com.vn/auth/oauth/check_token"
payload = {'token': post_save.token}
headers = {
'Authorization': 'Basic RGljdGlvbmFyeU1lZGlob21lOlJ4aXR6ZnZvaWFmZmNtb2l0ZW0='
}
response = requests.request(
"POST", url, headers=headers, data=payload)
data_output = json.loads(response.text)
data = data_output.get("user_name", None)
if data == None:
return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED, content={"message": "UNAUTHORIZED"})
datetime_now = datetime.now()
post_save.created_at = datetime_now.strftime("%m/%d/%y %H:%M:%S")
post_save = jsonable_encoder(post_save)
del post_save["token"]
post_save["username"] = data
new_post = await db["post_save"].insert_one(post_save)
created_post = await db["post_save"].find_one({"_id": new_post.inserted_id})
return JSONResponse(status_code=status.HTTP_201_CREATED, content=created_post)
except NameError:
return NameError
@post_save.get(
"/list_save_post_by_user", response_description="List save posts", response_model=SavePostModel
)
async def list_post(post_save: SavePost):
url = "https://sandboxapi.ebacsi.com.vn/auth/oauth/check_token"
payload = {'token': post_save.token}
headers = {
'Authorization': 'Basic RGljdGlvbmFyeU1lZGlob21lOlJ4aXR6ZnZvaWFmZmNtb2l0ZW0='
}
response = requests.request(
"POST", url, headers=headers, data=payload)
data_output = json.loads(response.text)
data = data_output.get("user_name", None)
if data == None:
return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED, content={"message": "UNAUTHORIZED"})
posts = await db["post_save"].find({"username": data, "is_active": "true"}).to_list(1000)
print(posts)
print(len(posts))
return posts

215
app/src/routers/routers.py
View File

@@ -1,3 +1,5 @@
from os import access
from urllib import response
from fastapi import (
APIRouter,
Depends,
@@ -16,71 +18,120 @@ from ..models.models import (
from ..dependecies import (
get_current_user,
authenticate_user,
authenticate_user_oauth2,
create_access_token,
get_password_hash
)
from ..settings import db, ACCESS_TOKEN_EXPIRE_MINUTES
import json
from typing import List
from datetime import datetime, timedelta
import requests
import re
from pydantic import BaseModel, Field
class LoginRequest(BaseModel):
username: str
password: str
grant_type: str
class TokenModel(BaseModel):
token: str = None
refresh_token: str = None
grant_type: str = None
router = APIRouter()
# ============= Creating path operations ==============
@router.post("/create_user", response_description="Add new user", response_model=UserModel)
async def create_user(user: UserModel, file: UploadFile = File(...)):
if re.match("admin|dev|simple mortal", user.role):
datetime_now = datetime.now()
user.created_at = datetime_now.strftime("%m/%d/%y %H:%M:%S")
user.password = get_password_hash(user.password)
user = jsonable_encoder(user)
file_location = f"../media/"
current_time = datetime_now.strftime("%H:%M:%S_%d-%m-%Y_")
file_save = file_location + current_time + file.filename
user.avatar = current_time + file.filename
with open(file_save, "wb+") as file_object:
file_object.write(file.file.read())
new_user = await db["users"].insert_one(user)
await db["users"].update_one({"_id": new_user.inserted_id}, {
"$rename": {"password": "hashed_pass"}})
created_user = await db["users"].find_one({"_id": new_user.inserted_id})
return JSONResponse(status_code=status.HTTP_201_CREATED, content=created_user)
raise HTTPException(status_code=406, detail="User role not acceptable")
# @router.post("/create_user", response_description="Add new user", response_model=UserModel)
# async def create_user(user: UserModel
# # , file: UploadFile = File(...)
# ):
# if re.match("admin|user|editor", user.role):
# datetime_now = datetime.now()
# user.created_at = datetime_now.strftime("%m/%d/%y %H:%M:%S")
# user.password = get_password_hash(user.password)
# user = jsonable_encoder(user)
# new_user = await db["users"].insert_one(user)
# await db["users"].update_one({"_id": new_user.inserted_id}, {
# "$rename": {"password": "hashed_pass"}})
# created_user = await db["users"].find_one({"_id": new_user.inserted_id})
# return JSONResponse(status_code=status.HTTP_201_CREATED, content=created_user)
# raise HTTPException(status_code=406, detail="User role not acceptable")
@router.post("/login")
async def login_for_access_token(body: LoginRequest):
print(body)
user = await authenticate_user(body.username, body.password)
print(body)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorect ID or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user["username"]}, expires_delta=access_token_expires
)
await db["users"].update_one({"username": body.username}, {"$set": {
"last_login": datetime.now().strftime("%m/%d/%y %H:%M:%S"),
"is_active": "true"
}})
url = "https://sandboxapi.ebacsi.com.vn/auth/oauth/token"
payload = {'username': body.username,
'password': body.password,
'grant_type': body.grant_type}
files = [
]
headers = {
'Authorization': 'Basic RGljdGlvbmFyeU1lZGlob21lOlJ4aXR6ZnZvaWFmZmNtb2l0ZW0=',
'Cookie': 'JSESSIONID=node0gmjiiq3ht7kv1gesg74t1pxsb20316.node0; XSRF-TOKEN=0976f6e0-814e-4be9-b6fa-b8d0c0896315'
}
response = requests.request(
"POST", url, headers=headers, data=payload, files=files)
access_token = json.loads(response.text)
url = "https://sandboxapi.ebacsi.com.vn/auth/oauth/check_token"
payload = {'token': access_token["access_token"]}
headers = {
'Authorization': 'Basic RGljdGlvbmFyeU1lZGlob21lOlJ4aXR6ZnZvaWFmZmNtb2l0ZW0='
}
response = requests.request(
"POST", url, headers=headers, data=payload)
data_output = json.loads(response.text)
data = data_output.get("user_name", None)
if data == None:
return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED, content={"message": "UNAUTHORIZED"})
user = json.loads(response.text)
access_token["authorities"] = user["authorities"]
return access_token
# @router.post("/token")
# async def login_for_access_token(body: OAuth2PasswordRequestForm = Depends()):
# url = "https://sandboxapi.ebacsi.com.vn/auth/oauth/token"
# payload = {'username': body.username,
# 'password': body.password,
# 'grant_type': body.grant_type}
# files = [
# ]
# headers = {
# 'Authorization': 'Basic RGljdGlvbmFyeU1lZGlob21lOlJ4aXR6ZnZvaWFmZmNtb2l0ZW0=',
# 'Cookie': 'JSESSIONID=node0gmjiiq3ht7kv1gesg74t1pxsb20316.node0; XSRF-TOKEN=0976f6e0-814e-4be9-b6fa-b8d0c0896315'
# }
# response = requests.request(
# "POST", url, headers=headers, data=payload, files=files)
# data = json.loads(response.text)
# del data["refresh_token"]
# del data["expires_in"]
# del data["scope"]
# del data["user_type"]
# del data["user_key"]
# return data
return {"access_token": access_token, "token_type": "bearer"}
@router.post("/token")
async def login_for_access_token_2(body: LoginRequest):
print(body)
user = await authenticate_user(body.username, body.password)
async def login_for_access_token_2(body: OAuth2PasswordRequestForm = Depends()):
user = await authenticate_user_oauth2(body.username, body.password)
print(body)
if not user:
raise HTTPException(
@@ -88,69 +139,99 @@ async def login_for_access_token_2(body: LoginRequest):
detail="Incorect ID or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user["username"]}, expires_delta=access_token_expires
)
await db["users"].update_one({"username": body.username}, {"$set": {
"last_login": datetime.now().strftime("%m/%d/%y %H:%M:%S"),
"is_active": "true"
}})
# access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
# access_token = create_access_token(
# data={"sub": user["username"]}, expires_delta=access_token_expires
# )
# await db["users"].update_one({"username": body.username}, {"$set": {
# "last_login": datetime.now().strftime("%m/%d/%y %H:%M:%S"),
# "is_active": "true"
# }})
return {"access_token": user["access_token"], "token_type": "bearer"}
return {"access_token": access_token, "token_type": "bearer"}
@router.get(
"/list", response_description="List all users", response_model=List[ShowUserModel]
)
async def list_users(current_user: ShowUserModel = Depends(get_current_user)):
async def list_users():
users = await db["users"].find().to_list(1000)
for user in users:
user["is_active"] = "false"
try:
last_login = datetime.strptime(user["last_login"], "%m/%d/%y %H:%M:%S")
last_login = datetime.strptime(
user["last_login"], "%m/%d/%y %H:%M:%S")
my_delta = datetime.now() - last_login
if my_delta <= timedelta(days=30):
user["is_active"] = "true"
except ValueError:
pass
return users
@router.post("/current", response_description="Current User")
async def current_user(token: TokenModel):
try:
url = "https://sandboxapi.ebacsi.com.vn/auth/oauth/check_token"
payload = {'token': token.token}
headers = {
'Authorization': 'Basic RGljdGlvbmFyeU1lZGlob21lOlJ4aXR6ZnZvaWFmZmNtb2l0ZW0='
}
response = requests.request(
"POST", url, headers=headers, data=payload)
data_output = json.loads(response.text)
data = data_output.get("user_name", None)
if data == None:
return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED, content={"message": "UNAUTHORIZED"})
else:
return json.loads(response.text)
except ValueError:
pass
@router.get("/current", response_description="Current User", response_model=ShowUserModel)
async def current_user(current_user: ShowUserModel = Depends(get_current_user)):
return current_user
# @router.get("/current", response_description="Current User", response_model=ShowUserModel)
# async def current_user(current_user: ShowUserModel = Depends(get_current_user)):
# return current_user
@router.post("/refresh_token", response_description="refresh token")
async def refresh_token(refresh_token: TokenModel):
url = "https://sandboxapi.ebacsi.com.vn/auth/oauth/token"
payload = {'refresh_token': refresh_token.refresh_token,
'grant_type': refresh_token.grant_type}
files = [
]
headers = {
'Authorization': 'Basic RGljdGlvbmFyeU1lZGlob21lOlJ4aXR6ZnZvaWFmZmNtb2l0ZW0=',
'Cookie': 'JSESSIONID=node0oxtulscdyhrr1ij9hfhpjl76825093.node0; XSRF-TOKEN=79789f05-27c4-4b2a-a0dc-4491894046ec'
}
response = requests.request(
"POST", url, headers=headers, data=payload, files=files)
return json.loads(response.text)
@router.put("/admin/{user_id}", response_description="Update a user", response_model=UpdateUserModel)
async def update_user(user_id: str, user: UpdateUserModel, current_user: UserModel = Depends(get_current_user)):
async def update_user(user_id: str, user: UpdateUserModel):
if current_user["role"] == "admin":
user = {k: v for k, v in user.dict().items() if v is not None}
if len(user) >= 1:
update_result = await db["users"].update_one({"_id": user_id}, {"$set": user})
if update_result.modified_count == 1:
if (
updated_user := await db["users"].find_one({"_id": user_id})
) is not None:
return updated_user
if (existing_user := await db["users"].find_one({"_id": user_id})) is not None:
return existing_user
raise HTTPException(status_code=404, detail=f"User {user_id} not found")
raise HTTPException(
status_code=404, detail=f"User {user_id} not found")
else:
raise HTTPException(status_code=403, detail=f"Not having sufficient rights to modify the content")
raise HTTPException(
status_code=403, detail=f"Not having sufficient rights to modify the content")
@router.delete("/delete_user/{user_id}", response_description="Delete a user")
async def delete_user(user_id: str, current_user: ShowUserModel = Depends(get_current_user)):
async def delete_user(user_id: str):
delete_result = await db["users"].delete_one({"_id": user_id})
if delete_result.deleted_count == 1:
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)

107
app/src/settings.py
View File

@@ -5,16 +5,115 @@ import os
import motor.motor_asyncio
# ================= Creating necessary variables ========================
#------------------ Token, authentication variables ---------------------
# ------------------ Token, authentication variables ---------------------
SECRET_KEY = "4ab5be85c8c56eecdd547f7831979be83de58a6768d10a314f54cda4e4d67ffe"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl="https://sandboxapi.ebacsi.com.vn/auth/oauth/token")
ROLE_ORG = [
{
"name": "ORG_USER",
"comment": "Bệnh nhân",
},
{
"name": "ORG_OPERATOR",
"comment": "CSKH hệ thống",
},
{
"name": "DOCTOR",
"comment": "Bác sĩ",
},
{
"name": "ORG_EXTERNAL",
"comment": "Tích hợp",
},
{
"name": "SITE_CASHIER",
"comment": "Thu ngân, thẩm định",
},
{
"name": "SITE_RECEIPTION",
"comment": "Lễ tân",
},
{
"name": "ORG_UTILS",
"comment": "SYSTEM",
},
{
"name": "SITE_ADMIN",
"comment": "Quản trị viên tại CSYT",
},
{
"name": "SITE_OPERATOR",
"comment": "CSKH tại CSYT",
},
{
"name": "SITE_INTEGRATION",
"comment": "Tài khoản tích hợp",
},
{
"name": "ORG_ADMIN",
"comment": "Quản trị hệ thống",
},
{
"name": "ORG_DICTIONARY",
"comment": "For Medihome Dictionary",
}
]
#----------------- Database variables (MongoDB) --------------------------
KHOA = [{
"name": "KHOA_NOI",
"comment": "Khoa Nội",
},
{
"name": "KHOA_NGOAI",
"comment": "Khoa Ngoại",
},
{
"name": "KHOA_PHU_SAN",
"comment": "Khoa Phụ sản",
},
{
"name": "KHOA_NHI",
"comment": "Khoa Nhi",
},
{
"name": "KHOA_TRUYEN_NHIEM",
"comment": "Khoa Truyền nhiễm",
},
{
"name": "KHOA_CAP_CUU",
"comment": "Khoa Cấp cứu",
},
{
"name": "KHOA_HOI_SUC",
"comment": "Khoa Hồi sức tích cực và chống độc",
},
{
"name": "KHOA_Y_HOC_CO_TRUYEN",
"comment": "Khoa Y học cổ truyền",
},
{
"name": "KHOA_U_BUOU",
"comment": "Khoa Ung bướu",
},
{
"name": "KHOA_Y_HOC_HAT_NHAN",
"comment": "Khoa Y học Hạt nhân",
},
{
"name": "KHOA_PHAU_THUA",
"comment": "Khoa Phẫu thuật - gây mê hồi sức",
},
{
"name": "KHOA_CHAN_DOAN_HINH_ANH",
"comment": "Khoa Chẩn đoán hình ảnh",
}]
# ----------------- Database variables (MongoDB) --------------------------
client = motor.motor_asyncio.AsyncIOMotorClient(os.environ["DB_URL"])
db = client.myTestDB
db = client.myTestDB