154 lines
4.0 KiB
Python
154 lines
4.0 KiB
Python
from datetime import datetime, time, timedelta
|
|
from fastapi import APIRouter, Depends, Header, HTTPException, status
|
|
from pydantic import BaseModel, EmailStr, Field
|
|
from typing import Dict, List, Optional, Set, Union
|
|
|
|
from ..lib_general import *
|
|
from app.config import settings
|
|
|
|
from app.db import *
|
|
|
|
|
|
#import logging
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class UserBase(BaseModel):
|
|
id_random: str = None # This should not be None. It is required.
|
|
account_id_random: str = None # This should not be None. It is required.
|
|
username: str = Field(None, example='New.User', min_length=3, max_length=100)
|
|
name: Optional[str] = None
|
|
email: EmailStr
|
|
email_verified: Optional[bool] = None
|
|
enable: Optional[bool] = None
|
|
enable_from: Optional[datetime] = None
|
|
enable_to: Optional[datetime] = None
|
|
super: Optional[bool] = None
|
|
manager: Optional[bool] = None
|
|
administrator: Optional[bool] = None
|
|
verified: Optional[bool] = None
|
|
|
|
|
|
class UserIn(UserBase):
|
|
password: str = Field(None, example='My Difficult Password!', min_length=10)
|
|
|
|
|
|
class UserOut(UserBase):
|
|
password_set_on: Optional[datetime] = None
|
|
password_reset_token: Optional[str] = None
|
|
password_reset_expire_on: Optional[datetime] = None
|
|
logged_in_on: Optional[datetime] = None
|
|
last_activity_on: Optional[datetime] = None
|
|
created_on: datetime
|
|
update_on: Optional[datetime] = None
|
|
|
|
|
|
class UserInDB(UserBase):
|
|
hashed_password: str
|
|
password_set_on: Optional[datetime] = None
|
|
password_reset_token: Optional[str] = None
|
|
password_reset_expire_on: Optional[datetime] = None
|
|
logged_in_on: Optional[datetime] = None
|
|
last_activity_on: Optional[datetime] = None
|
|
|
|
|
|
@router.post(
|
|
"/",
|
|
response_model=UserOut,
|
|
summary='Create a new user account',
|
|
status_code=status.HTTP_201_CREATED
|
|
)
|
|
async def create_user(user: UserIn, x_account_id: str = Header(...)):
|
|
"""
|
|
Create a new user account
|
|
"""
|
|
|
|
user = {}
|
|
user['account_id_random'] = x_account_id
|
|
user['username'] = 'Scott.Idem'
|
|
user['name'] = 'Scott Idem'
|
|
user['email'] = 'Scott.Idem@oneskyit.com'
|
|
|
|
return user
|
|
|
|
|
|
#@router.patch('/{id_random}', response_model=UserOut, dependencies=[Depends(get_account_header)])
|
|
#async def update_user(id_random: str, user: UserIn, x_account_id: str = Header(...)):
|
|
#async def update_user(id_random: str, user: UserIn):
|
|
@router.patch(
|
|
'/{id_random}',
|
|
response_model=UserOut,
|
|
summary='Update a user account'
|
|
)
|
|
async def update_user(id_random: str, user: UserIn, x_account_id: str = Depends(get_account_header)):
|
|
"""
|
|
Update a user account
|
|
"""
|
|
|
|
user = {}
|
|
user['id_random'] = id_random
|
|
user['account_id_random'] = x_account_id
|
|
user['username'] = 'Scott.Idem'
|
|
user['name'] = 'Scott Idem'
|
|
user['email'] = 'Scott.Idem@oneskyit.com'
|
|
user['created_on'] = datetime.now()
|
|
user['super'] = True
|
|
|
|
return user
|
|
|
|
|
|
@router.delete('/{id_random}', response_model=bool)
|
|
async def delete_user(id_random: str, x_account_id: str = Depends(get_account_header)):
|
|
"""
|
|
Delete a user account
|
|
"""
|
|
|
|
return True
|
|
return False
|
|
|
|
|
|
@router.get('/', response_model=List[UserOut])
|
|
@router.get('/list_all', response_model=List[UserOut])
|
|
async def list_users():
|
|
"""
|
|
Get a list of users
|
|
"""
|
|
|
|
print(settings.APP_NAME)
|
|
|
|
users = [{'username': 'test.user.1'}, {'username': 'test.user.2'}, {'username': 'Scott.Idem'}]
|
|
|
|
|
|
print('Getting all users...')
|
|
|
|
sql = """
|
|
SELECT *
|
|
FROM `user`
|
|
WHERE id=1
|
|
"""
|
|
|
|
records = sql_select(sql=sql, as_list=True)
|
|
|
|
#records = sql_select(table_name='user')
|
|
|
|
|
|
if records:
|
|
print('Got the user list')
|
|
return records
|
|
else:
|
|
print('No user records found')
|
|
raise HTTPException(status_code=404)
|
|
|
|
|
|
@router.get('/{username}')
|
|
async def get_user_username(username: str, x_account_id: str = Header(...)):
|
|
return {'username': username}
|
|
|
|
|
|
#@router.get('/me')
|
|
#async def get_user_current():
|
|
#user_out: UserOut
|
|
|
|
#return {'username': 'test.user'}
|