150 lines
4.5 KiB
Python
150 lines
4.5 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 ..log import *
|
|
from app.config import settings
|
|
from app.db import *
|
|
from .user_models import *
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post(
|
|
"/",
|
|
response_model=UserOut,
|
|
response_model_exclude_unset=True,
|
|
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 = dict(user)
|
|
table_name = 'user'
|
|
|
|
# Look up the user['account_id_random'] and match to a record ID from Redis
|
|
if account_id := redis_lookup_id_random(table_name='account', record_id_random=x_account_id):
|
|
user['account_id'] = account_id
|
|
user.pop('account_id_random')
|
|
else:
|
|
print('Something went wrong with the id_random lookup.')
|
|
raise HTTPException(status_code=500)
|
|
|
|
if result := sql_insert(table_name=table_name, record=user, id_random_length=16):
|
|
print(type(result))
|
|
if type(result) == int: # isinstance(result, int):
|
|
# Select the new record to return as a response.
|
|
if new_user := dict(sql_select(table_name=table_name, record_id=result)):
|
|
return new_user
|
|
else:
|
|
print('New user record was not found.')
|
|
raise HTTPException(status_code=400)
|
|
else:
|
|
print('There is likely a duplicate record. A new record was not created.')
|
|
raise HTTPException(status_code=400)
|
|
else:
|
|
print('No user record was not created')
|
|
raise HTTPException(status_code=400)
|
|
|
|
|
|
#@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
|
|
"""
|
|
log.setLevel(logging.DEBUG)
|
|
log.debug(locals())
|
|
|
|
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
|
|
"""
|
|
log.setLevel(logging.DEBUG)
|
|
log.debug(locals())
|
|
|
|
return True
|
|
return False
|
|
|
|
|
|
@router.get('/', response_model=List[UserOut])
|
|
@router.get('/list_all', response_model=List[UserOut])
|
|
async def list_users(x_account: str = Depends(get_account_header)):
|
|
"""
|
|
Get a list of users
|
|
"""
|
|
log.setLevel(logging.DEBUG)
|
|
log.debug(locals())
|
|
|
|
if x_account['id']:
|
|
log.info('The x-account-id was given and is not empty...')
|
|
sql = """
|
|
SELECT *
|
|
FROM `user`
|
|
WHERE account_id = :account_id
|
|
"""
|
|
records = sql_select(table_name='user', field_name='account_id', field_value=x_account['id'], as_list=True)
|
|
elif x_account['id'] is None:
|
|
log.info('The x-account-id was given, but is empty...')
|
|
sql = """
|
|
SELECT *
|
|
FROM `user`
|
|
"""
|
|
records = sql_select(table_name='user', as_list=True)
|
|
|
|
if records:
|
|
log.info('Returning a user list...')
|
|
return records
|
|
else:
|
|
log.info('No user records found...')
|
|
raise HTTPException(status_code=404)
|
|
|
|
|
|
@router.get('/{username}')
|
|
async def get_user_username(username: str, x_account: str = Depends(get_account_header)):
|
|
log.setLevel(logging.DEBUG)
|
|
log.debug(locals())
|
|
|
|
data = {}
|
|
data['username'] = username
|
|
|
|
if x_account['id']:
|
|
sql = """
|
|
SELECT *
|
|
FROM `user`
|
|
WHERE account_id = :account_id AND username=:username
|
|
"""
|
|
data['account_id'] = x_account['id']
|
|
elif x_account['id'] is None:
|
|
sql = """
|
|
SELECT *
|
|
FROM `user`
|
|
WHERE (account_id IS NULL OR account_id = "") AND username=:username
|
|
"""
|
|
record = sql_select(sql=sql, data=data)
|
|
|
|
return record
|