Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added person lookup by email and email auth url key.

Browse Source
This commit is contained in:
Scott Idem
2021-12-02 17:57:01 -05:00
parent 2c0af0a321
commit 630739aae6
7 changed files with 265 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
app/lib_general.py
View File

@@ -1,7 +1,15 @@
from __future__ import annotations
import datetime, jwt, os, pandas, pathlib, pytz, redis, time
import datetime, html2text, jwt, os, pandas, pathlib, pytz, redis, time
from passlib.hash import argon2
# Import smtplib for the actual sending function
import smtplib, ssl
# Import the email package modules needed
from email.message import EmailMessage
from email.headerregistry import Address
from email.utils import make_msgid
from fastapi import APIRouter, Depends, Header, HTTPException, Response, status
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
@@ -163,3 +171,90 @@ def create_export_file(
return tmp_file_path # True
# ### END ### API Lib General ### create_export() ###
# ### BEGIN ### API Lib General ### send_email() ###
# Updated 2021-12-02
@logger_reset
def send_email(
from_email:str,
to_email: str,
subject: str,
body_html: str,
from_name: str = '',
reply_to_email: str = '',
reply_to_name: str = '',
to_name: str = '',
cc_email: str = '',
cc_name: str = '',
bcc_email: str = '',
bcc_name: str = '',
body_text: str = '',
):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
message = EmailMessage()
if subject:
message['Subject'] = subject
else:
return False
if from_email:
#message['From'] = Address(display_name=from_name, username=from_email.split('@')[0], domain=from_email.split('@')[1])
message['From'] = Address(display_name=from_name, addr_spec=from_email)
else:
return False
if reply_to_email:
message['Reply-To'] = Address(display_name=reply_to_name, addr_spec=reply_to_email)
if to_email:
message['To'] = Address(display_name=to_name, addr_spec=to_email)
else:
return False
if cc_email:
message['Cc'] = Address(display_name=cc_name, addr_spec=cc_email)
if bcc_email:
message['Bcc'] = Address(display_name=bcc_name, addr_spec=bcc_email)
html_version = """\
<html>
<body>
<div>
"""+body_html+"""
</div>
</body>
</html>
"""
if body_text:
text_version = body_text
else:
text_version = html2text.html2text(html_version)
message.set_content(text_version)
message.add_alternative(html_version, subtype='html')
log.info('Sending email...')
log.debug(settings.SMTP)
log.info(f'Subject: {subject}')
log.info(f'From: {from_email} Reply To: {reply_to_email} To: {to_email} CC: {cc_email} BCC: {bcc_email}')
log.debug('Message:')
log.debug(message.as_string())
log.info('Creating SMTP SSL connection...')
context = ssl.create_default_context()
log.info('SMTP configuration, connect, and send')
try:
with smtplib.SMTP_SSL(settings.SMTP['server_name'], settings.SMTP['server_port'], context=context) as server:
server.login(settings.SMTP['username'], settings.SMTP['password'])
server.send_message(message)
log.info('Email sent')
return True
except:
#except SMTPException:
log.error('Error: unable to send email')
return False
# ### END ### API Lib General ### send_email() ###

4
app/methods/event_methods.py
View File

@@ -13,7 +13,7 @@ from app.methods.event_cfg_methods import create_update_event_cfg_obj_v4, load_e
from app.methods.event_location_methods import get_event_location_rec_list, load_event_location_obj
from app.methods.event_session_methods import get_event_session_rec_list, load_event_session_obj
from app.methods.person_methods import create_update_person_obj_v4b, load_person_obj
from app.methods.user_methods import create_user_obj, load_user_obj, update_user_obj
# from app.methods.user_methods import create_user_obj, load_user_obj, update_user_obj
from app.models.common_field_schema import default_num_bytes
from app.models.event_models import Event_Base
@@ -736,6 +736,7 @@ def update_event_obj(
if event_obj.user_id and event_obj.user:
event_outline['user_id'] = None
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
from app.methods.user_methods import update_user_obj
user_id = event_obj.user_id
user_obj_up = event_obj.user
log.debug(user_id)
@@ -751,6 +752,7 @@ def update_event_obj(
return False
elif event_obj.user and not event_obj.user.id:
# NOTE: This will blindly create a new user even if there was one associated but the event.user_id was not found.
from app.methods.user_methods import create_user_obj
user_obj_in = event_obj.user
log.debug(user_obj_in)
if user_obj_in_result := create_user_obj(account_id=account_id, user_obj_new=user_obj_in):

5
app/methods/event_person_methods.py
View File

@@ -18,7 +18,7 @@ from app.methods.event_person_profile_methods import load_event_person_profile_o
# from app.methods.event_session_methods import load_event_session_obj
# from app.methods.event_track_methods import load_event_track_obj
from app.methods.person_methods import create_person_obj_v3, load_person_obj, update_person_obj
from app.methods.user_methods import create_user_obj, load_user_obj, update_user_obj
# from app.methods.user_methods import create_user_obj, load_user_obj, update_user_obj
from app.models.common_field_schema import default_num_bytes
from app.models.event_person_models import Event_Person_New_Base, Event_Person_Base
@@ -170,6 +170,7 @@ def load_event_person_obj(
if inc_user:
log.info('Need to include user data...')
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
from app.methods.user_methods import load_user_obj
if user_obj := load_user_obj(
user_id = user_id
):
@@ -876,6 +877,7 @@ def update_event_person_obj(
if event_person_obj_up.user_id and event_person_obj_up.user:
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
from app.methods.user_methods import update_user_obj
user_id = event_person_obj_up.user_id
user_obj_up = event_person_obj_up.user
log.debug(user_id)
@@ -891,6 +893,7 @@ def update_event_person_obj(
return False
elif event_person_obj_up.user and not event_person_obj_up.user.id:
# NOTE: This will blindly create a new user even if there was one associated but the event_person.user_id was not found.
from app.methods.user_methods import create_user_obj
user_obj_in = event_person_obj_up.user
log.debug(user_obj_in)
if user_obj_in_result := create_user_obj(account_id=account_id, user_obj_new=user_obj_in):

123
app/methods/user_methods.py
View File

@@ -1,12 +1,13 @@
from __future__ import annotations
import datetime, random
import datetime, random, secrets
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from app.db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from app.lib_general import log, logging
from app.lib_general import log, logging, send_email
# from app.methods.account_methods import load_account_cfg_obj
from app.methods.contact_methods import load_contact_obj, update_contact_obj
# from app.methods.event_methods import get_event_rec_list
from app.methods.order_methods import load_order_obj, get_order_rec_list
@@ -15,6 +16,7 @@ from app.methods.person_methods import create_person_obj_v3, load_person_obj, up
from app.methods.post_methods import get_post_rec_list, load_post_obj
from app.methods.user_role_methods import get_user_role_rec_list, load_user_role_obj
from app.models.common_field_schema import default_num_bytes
from app.models.user_models import User_Base, User_New_Base, User_Out_Base
@@ -482,3 +484,120 @@ def get_user_rec_list(
return user_rec_li
# ### END ### API User Methods ### get_user_rec_list() ###
# ### BEGIN ### User Methods ### email_user_auth_key_url() ###
# This emails the actual one time use sign in URL for a user.
# Updated 2021-12-02
def email_user_auth_key_url(
account_id: int|str,
user_id: int|str,
root_url: str,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
if user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass
else: return False
if user_obj := load_user_obj(
user_id = user_id,
):
log.info('User object loaded')
if user_obj.enable and user_obj.allow_auth_key:
new_auth_key = secrets.token_urlsafe(default_num_bytes)
update_user_data = {}
update_user_data['auth_key'] = new_auth_key
if user_rec_update_result := sql_update(
table_name = 'user',
record_id = user_id,
data = update_user_data
):
log.info('The user record was updated with a new auth_key')
else:
log.warning('The user record was not updated with a new auth_key')
return False
else:
log.warning('The user record was not enabled or auth_key is not allowed')
return False
else: return False
log.debug(user_obj)
from app.methods.account_cfg_methods import load_account_cfg_obj
if account_cfg := load_account_cfg_obj(
account_id = account_id,
):
log.info('Account config loaded')
else: return False
log.debug(account_cfg)
user_id_random = user_obj.id_random # NOTE: Not user_id_random because of alias
from_email = account_cfg.default_no_reply_email
from_name = account_cfg.default_no_reply_name
to_name = user_obj.name
to_email = user_obj.email
bcc_email = account_cfg.confirm_email
bcc_name = account_cfg.confirm_name
help_tech_email = account_cfg.help_tech_email
help_tech_name = account_cfg.help_tech_name
account_short_name = account_cfg.account_short_name
username = user_obj.username
enable = user_obj.enable
if enable_from := user_obj.enable_from:
# enable_from_datetime = datetime.datetime.fromisoformat(enable_from).replace(tzinfo=datetime.timezone.utc)
enable_from_datetime = enable_from.replace(tzinfo=datetime.timezone.utc)
enable_from_datetime = enable_from
enable_from_str = enable_from_datetime.strftime('%A, %B %-d, %Y %-I:%M %p %Z')
else: enable_from_str = '-- Not Set --'
if enable_to := user_obj.enable_to:
# enable_to_datetime = datetime.datetime.fromisoformat(enable_to).replace(tzinfo=datetime.timezone.utc)
enable_to_datetime = enable_to.replace(tzinfo=datetime.timezone.utc)
enable_to_str = enable_to_datetime.strftime('%A, %B %-d, %Y %-I:%M %p %Z')
else: enable_to_str = '-- Not Set --'
auth_key = user_obj.auth_key
user_login_url = f'{root_url}user/login?username={username}'
user_login_auth_key_url = f'{root_url}?user_id={user_id_random}&auth_key={new_auth_key}'
subject = f'{account_short_name}: One Time Use Sign In Link ({new_auth_key})'
body_html = f"""
<p>{to_name},</p>
<p>If you did not request this sign in link, please delete this email. It is suggested that you delete this email after the sign in link has been used or if a new link has been requested.</p>
<p>The link below can only be used to sign in once. If you would like to sign in again using this method, you must <a href="{user_login_url}">request a new sign in link</a>. If you request multiple links, only the newest link will sign you in.</p>
<p><strong><a href="{user_login_auth_key_url}" style="appearance: button; display: inline-block; text-align: center; text-decoration: none; padding: .2rem .4rem; border: solid thin gray; border-radius: .2rem; background-color: lightyellow; color: black; font-size: larger;">Click to Sign In With One Time Use Link</a></strong></p>
<p>Or copy and paste the link:<br>
<strong style="background-color: lightyellow; color: black; font-size: larger;"><a href="{user_login_auth_key_url}">{user_login_auth_key_url}</a></strong></p>
<p>Your username is: {username}<br>
User account enabled: {enable}<br>
User account enabled from: {enable_from_str}<br>
User account enabled to: {enable_to_str}<br>
Current one time use auth key for user account: {new_auth_key}</p>
<p>If you have questions about this email or trouble with this one time use link, you can email <a href="mailto:{help_tech_email}">{help_tech_name} ({help_tech_email})</a>.</p>
<p>Thank you!</p>
"""
if send_email(from_email=from_email, from_name=from_name, to_email=to_email, to_name=to_name, bcc_email=bcc_email, bcc_name=bcc_name, subject=subject, body_text=None, body_html=body_html):
log.info(f'An email with a one time use sign in link was sent to {to_email}.')
return True
else:
log.info(f'An email with a one time use sign in link was not sent to {to_email}.')
return False
# ### END ### User ### email_user_auth_key_url() ###

3
app/models/account_cfg_models.py
View File

@@ -117,7 +117,7 @@ class Account_Cfg_Base(BaseModel):
@validator('account_id', always=True)
def account_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.setLevel(logging.DEBUG)
log.debug(locals())
if values['account_id_random']:
@@ -126,6 +126,7 @@ class Account_Cfg_Base(BaseModel):
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields
#Account_Cfg_Base.update_forward_refs()

4
app/routers/person.py
View File

@@ -394,9 +394,9 @@ async def lookup_email(
if account_id := redis_lookup_id_random(record_id_random=x_account_id, table_name='account'): pass
else: return mk_resp(data=None, status_code=404, response=response)
import time
# import time
time.sleep(1)
# time.sleep(1)
if person_rec_list_result := get_person_rec_list(
#account_id = account_id,

38
app/routers/user.py
View File

@@ -10,7 +10,7 @@ from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select,
from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
from app.methods.order_methods import get_order_rec_list, load_order_obj
from app.methods.user_methods import create_user_obj, get_user_rec_list, load_user_obj
from app.methods.user_methods import create_user_obj, email_user_auth_key_url, get_user_rec_list, load_user_obj
from app.models.common_field_schema import default_num_bytes
from app.models.response_models import Resp_Body_Base, mk_resp
@@ -806,6 +806,42 @@ async def lookup_username(
return mk_resp(data=data, response=response)
# ### BEGIN ### API User ### email_auth_key_url() ###
# Updated 2021-12-02
# @router.get('/user/email_auth_key_url', response_model=Resp_Body_Base)
@router.get('/user/{user_id}/email_auth_key_url', response_model=Resp_Body_Base)
async def email_auth_key_url(
user_id: Optional[str] = Query(None, min_length=11, max_length=22),
root_url: Optional[str] = Query(None, min_length=10, max_length=100), # Absolute min = 7
x_account_id: Optional[str] = Header(..., ),
return_obj: bool = False,
by_alias: bool = True,
exclude_unset: bool = True,
response: Response = Response,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if account_id := redis_lookup_id_random(record_id_random=x_account_id, table_name='account'): pass
else: return mk_resp(data=False, status_code=404, response=response) # Not Found
if user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass
else: return mk_resp(data=False, status_code=404, response=response) # Not Found
if result := email_user_auth_key_url(
account_id = account_id,
user_id = user_id,
root_url = root_url,
):
log.info('Email with auth key log in URL was sent.')
return mk_resp(data=True, response=response)
else:
log.warning('Email with auth key log in URL was not sent.')
return mk_resp(data=False, status_code=500, response=response)
# ### END ### API User ### email_auth_key_url() ###
# ### BEGIN ### API User ### get_user_obj() ###
# Working well as of 2021-06-25. Using as a template for other routes.
@router.get('/user/{user_id}', response_model=Resp_Body_Base)