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

Working on event_person and session proposals

Browse Source
This commit is contained in:
Scott Idem
2021-05-26 17:42:18 -04:00
parent 3afb81d801
commit 8f6589cf1c
14 changed files with 634 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
app/models/address_methods.py Normal file
View File

@@ -0,0 +1,33 @@
from __future__ import annotations
import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from ..lib_general import *
from ..db_sql import redis_lookup_id_random, sql_insert, sql_select
from .address_model import Address_Base
# ### BEGIN ### API Address Methods ### create_address_obj() ###
def create_address_obj(address_obj_new:Address_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not address_obj_new:
return False
address_obj_data = address_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'created_on', 'updated_on'})
if address_obj_in_result := sql_insert(data=address_obj_data, table_name='address', rm_id_random=True, id_random_length=8): pass
else:
return False
log.debug(address_obj_in_result)
address_id = address_obj_in_result
log.debug(f'Returning the new address_id: {address_id}')
return address_id
# ### END ### API Address Methods ### create_address_obj() ###

1
app/models/common_field_schema.py
View File

@@ -34,6 +34,7 @@ base_fields['event_abstract_id_random'] = xxx_id_random_field_schema
base_fields['event_badge_id_random'] = xxx_id_random_field_schema
base_fields['event_device_id_random'] = xxx_id_random_field_schema
base_fields['event_location_id_random'] = xxx_id_random_field_schema
base_fields['event_person_id_random'] = xxx_id_random_field_schema
base_fields['event_presentation_id_random'] = xxx_id_random_field_schema
base_fields['event_presenter_id_random'] = xxx_id_random_field_schema
base_fields['event_registration_id_random'] = xxx_id_random_field_schema

53
app/models/contact_methods.py Normal file
View File

@@ -0,0 +1,53 @@
from __future__ import annotations
import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from ..lib_general import *
from ..db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from .contact_model import Contact_Base
from ..models.address_methods import create_address_obj
# ### BEGIN ### API Contact Methods ### create_contact_obj() ###
def create_contact_obj(contact_obj_new:Contact_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not contact_obj_new:
return False
contact_obj_data = contact_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'address', 'created_on', 'updated_on'})
if contact_obj_in_result := sql_insert(data=contact_obj_data, table_name='contact', rm_id_random=True, id_random_length=8): pass
else:
return False
log.debug(contact_obj_in_result)
contact_id = contact_obj_in_result
if contact_obj_new.address:
address_obj_new = contact_obj_new.address
address_obj_new.for_type = 'contact'
address_obj_new.for_id = contact_id
create_address_obj_result = create_address_obj(address_obj_new=address_obj_new)
if isinstance(create_address_obj_result, int):
address_id = create_address_obj_result
# Need to update the contact with the new address_id
contact_obj_up = {}
contact_obj_up['id'] = contact_id
contact_obj_up['address_id'] = address_id
if contact_obj_up_result := sql_update(data=contact_obj_up, table_name='contact'): pass
else:
return False
log.debug(contact_obj_up_result)
else:
log.debug(f'No address_id was returned when tyring to create_address_obj(): {create_address_obj_result}')
address_id = None
log.debug(f'Returning the new contact_id: {contact_id}')
return contact_id
# ### END ### API Contact Methods ### create_contact_obj() ###

35
app/models/event_person_methods.py Normal file
View File

@@ -0,0 +1,35 @@
from __future__ import annotations
import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from ..lib_general import *
from ..db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from .event_person_model import Event_Person_New_Base, Event_Person_Base
# ### BEGIN ### API Event Person Methods ### create_event_person_obj() ###
def create_event_person_obj(event_person_obj_new:Event_Person_Base):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not event_person_obj_new:
return False
event_person_obj_data = event_person_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'created_on', 'updated_on'})
log.debug(event_person_obj_data)
if event_person_obj_in_result := sql_insert(data=event_person_obj_data, table_name='event_person', rm_id_random=True, id_random_length=8): pass
else:
return False
log.setLevel(logging.DEBUG)
log.debug(event_person_obj_in_result)
event_person_id = event_person_obj_in_result
log.debug(f'Returning the new event_person_id: {event_person_id}')
return event_person_id
# ### END ### API Event Person Methods ### create_event_person_obj() ###

209
app/models/event_person_model.py Normal file
View File

@@ -0,0 +1,209 @@
from __future__ import annotations
import datetime, hashlib, logging, os, pytz, redis, secrets
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator
from ..db_sql import redis_lookup_id_random
from ..lib_general import *
from .common_field_schema import base_fields, default_num_bytes
class Event_Person_New_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['event_person_id_random'],
alias='event_person_id_random',
default_factory=lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
#alias='event_person_id'
)
account_id_random: Optional[str]
account_id: Optional[int]
event_id_random: Optional[str]
event_id: Optional[int]
person_given_name: Optional[str]
person_family_name: Optional[str]
person_full_name: Optional[str]
organization_name: Optional[str]
email: Optional[str]
#new_password: str = Field(default_factory=lambda:secrets.token_urlsafe(default_num_bytes))
#password: Optional[str]
new_password: Optional[str]
notes: Optional[str]
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('event_person_id_random', always=True)
def event_person_id_random_copy(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']:
return values['id_random']
return None
@validator('id', always=True)
def event_person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='event_person')
return None
@validator('account_id', always=True)
def account_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['account_id_random']:
return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account')
return None
@validator('event_id', always=True)
def event_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['event_id_random']:
return redis_lookup_id_random(record_id_random=values['event_id_random'], table_name='event')
return None
# @validator('password', always=True)
# def hash_new_password(cls, v, values, **kwargs):
# log.setLevel(logging.WARNING)
# log.debug(locals())
# if values['new_password']:
# return secure_hash_string(string=values['new_password'])
# return None
class Config:
underscore_attrs_are_private = True
fields = base_fields
class Event_Person_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['event_person_id_random'],
alias='event_person_id_random',
default_factory=lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
#alias='event_person_id'
)
account_id_random: Optional[str]
account_id: Optional[int]
event_id_random: Optional[str]
event_id: Optional[int]
event_registration_id_random: Optional[str]
event_registration_id: Optional[int]
event_badge_id_random: Optional[str]
event_badge_id: Optional[int]
person_id_random: Optional[str]
person_id: Optional[int]
user_id_random: Optional[str]
user_id: Optional[int]
priority: Optional[bool]
sort: Optional[int]
group: Optional[str]
notes: Optional[str]
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('event_person_id_random', always=True)
def event_person_id_random_copy(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']:
return values['id_random']
return None
@validator('id', always=True)
def event_person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='event_person')
return None
@validator('account_id', always=True)
def account_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['account_id_random']:
return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account')
return None
@validator('event_id', always=True)
def event_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['event_id_random']:
return redis_lookup_id_random(record_id_random=values['event_id_random'], table_name='event')
return None
@validator('event_registration_id', always=True)
def event_registration_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['event_registration_id_random']:
return redis_lookup_id_random(record_id_random=values['event_registration_id_random'], table_name='event_registration')
return None
@validator('event_badge_id', always=True)
def event_badge_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['event_badge_id_random']:
return redis_lookup_id_random(record_id_random=values['event_badge_id_random'], table_name='event_badge')
return None
@validator('person_id', always=True)
def person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['person_id_random']:
return redis_lookup_id_random(record_id_random=values['person_id_random'], table_name='person')
return None
@validator('user_id', always=True)
def user_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['user_id_random']:
return redis_lookup_id_random(record_id_random=values['user_id_random'], table_name='user')
return None
class Config:
underscore_attrs_are_private = True
fields = base_fields

8
app/models/membership_methods.py
View File

@@ -15,7 +15,7 @@ from .membership_model import Membership_Base
#from .user_model import User_Base
# ### BEGIN ### API Membership Model ### save_membership_obj() ###
# ### BEGIN ### API Membership Methods ### save_membership_obj() ###
def save_membership_obj(order_obj_new:Membership_Base=None):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -159,10 +159,10 @@ def save_membership_obj(order_obj_new:Membership_Base=None):
else: return False
log.debug(order_line_obj_resp)
return order_id
# ### END ### API Membership Model ### save_membership_obj() ###
# ### END ### API Membership Methods ### save_membership_obj() ###
# ### BEGIN ### API Membership Model ### load_membership_obj() ###
# ### BEGIN ### API Membership Methods ### load_membership_obj() ###
def load_membership_obj(membership_id:int|str, inc_membership_profile:bool=False, inc_membership_cfg:bool=False, inc_extended_profile:bool=False, inc_person:bool=False, inc_user:bool=False) -> Membership_Base:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -223,4 +223,4 @@ def load_membership_obj(membership_id:int|str, inc_membership_profile:bool=False
log.error(e.json())
return membership_obj
# ### END ### API Membership Model ### load_membership_obj() ###
# ### END ### API Membership Methods ### load_membership_obj() ###

8
app/models/order_methods.py
View File

@@ -12,7 +12,7 @@ from .order_model import Order_Base
#from .user_model import User_Base
# ### BEGIN ### API Order Model ### save_order_obj() ###
# ### BEGIN ### API Order Methods ### save_order_obj() ###
def save_order_obj(order_obj_new:Order_Base, repl_order_line_li:bool=False):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -153,10 +153,10 @@ def save_order_obj(order_obj_new:Order_Base, repl_order_line_li:bool=False):
else: return False
log.debug(order_line_obj_resp)
return order_id
# ### END ### API Order Model ### save_order_obj() ###
# ### END ### API Order Methods ### save_order_obj() ###
# ### BEGIN ### API Order Model ### load_order_obj() ###
# ### BEGIN ### API Order Methods ### load_order_obj() ###
def load_order_obj(order_id:int|str, inc_order_line_li:bool=False, inc_order_cfg:bool=False, inc_person_obj:bool=False, inc_user_obj:bool=False):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -201,4 +201,4 @@ def load_order_obj(order_id:int|str, inc_order_line_li:bool=False, inc_order_cfg
log.error(e.json())
return order_obj
# ### END ### API Order Model ### load_order_obj() ###
# ### END ### API Order Methods ### load_order_obj() ###

49
app/models/person_methods.py
View File

@@ -5,14 +5,59 @@ from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from ..lib_general import *
from ..db_sql import redis_lookup_id_random, sql_select
from ..db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from .person_model import Person_Base
from ..models.contact_methods import create_contact_obj
# ### BEGIN ### API Person Methods ### create_person_obj() ###
def create_person_obj(person_obj_new:Person_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not person_obj_new:
return False
person_obj_data = person_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'contact', 'organization', 'created_on', 'updated_on'})
if person_obj_in_result := sql_insert(data=person_obj_data, table_name='person', rm_id_random=True, id_random_length=8): pass
else:
return False
log.setLevel(logging.DEBUG)
log.debug(person_obj_in_result)
person_id = person_obj_in_result
if person_obj_new.contact:
contact_obj_new = person_obj_new.contact
contact_obj_new.for_type = 'person'
contact_obj_new.for_id = person_id
create_contact_obj_result = create_contact_obj(contact_obj_new=contact_obj_new)
if isinstance(create_contact_obj_result, int):
contact_id = create_contact_obj_result
log.debug(f'Update person with new contact_id: {contact_id}')
# Need to update the person with the new contact_id
person_obj_up = {}
person_obj_up['id'] = person_id
person_obj_up['contact_id'] = contact_id
if person_obj_up_result := sql_update(data=person_obj_up, table_name='person'): pass
else:
return False
log.debug(person_obj_up_result)
else:
log.debug(f'No contact_id was returned when tyring to create_contact_obj(): {create_contact_obj_result}')
contact_id = None
log.debug(f'Returning the new person_id: {person_id}')
return person_id
# ### END ### API Person Methods ### create_person_obj() ###
# ### BEGIN ### API Person Methods ### load_person_obj() ###
def load_person_obj(person_id:int|str, inc_contact:bool=False, inc_address:bool=False, inc_organization:bool=False) -> Person_Base:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass

11
app/models/person_model.py
View File

@@ -36,6 +36,8 @@ class Person_Base(BaseModel):
id: Optional[int] = Field(
#alias='person_id'
)
account_id_random: Optional[str]
account_id: Optional[int]
user_id_random: Optional[str]
user_id: Optional[int]
organization_id_random: Optional[str]
@@ -86,6 +88,15 @@ class Person_Base(BaseModel):
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='person')
return None
@validator('account_id', always=True)
def account_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['account_id_random']:
return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account')
return None
@validator('user_id', always=True)
def user_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)

56
app/models/user_methods.py
View File

@@ -5,12 +5,66 @@ from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from ..lib_general import *
from ..db_sql import redis_lookup_id_random, sql_select
from ..db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from .user_model import User_Base, User_Out_Base, User_New_Base
from .user_role_model import User_Role_Base
# ### BEGIN ### API User Methods ### create_user_obj() ###
def create_user_obj(user_obj_new:User_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not user_obj_new:
return False
# user_obj_data = user_obj_new.dict(by_alias=False, exclude_defaults=False, include={'password'}, exclude={'new_password'})
# log.debug(user_obj_data)
user_obj_data = user_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'new_password'})
log.debug(user_obj_data)
user_obj_data['password'] = user_obj_new.password # There has to be a better way to do this??? It thinks "password" is unset and so is excluded?
log.debug(user_obj_data)
log.info('Checking if the username is already in use for the account...')
sql_select_user = f"""
SELECT user.id, user.id_random, user.name, user.email
FROM `user` AS user
WHERE user.account_id = :account_id and user.username = :username
"""
if sql_select_result := sql_select(sql=sql_select_user, data=user_obj_data, rm_id_random=True):
if isinstance(sql_select_result, list):
log.exception(f"Multiple user accounts already exists with this username ({user_obj_data['username']}). The database needs to be checked.")
return False
log.info('A user account already exists with this username. Updating instead of inserting.')
user_id = sql_select_result.get('id', None)
user_obj_data['id'] = user_id
if user_obj_up_result := sql_update(data=user_obj_data, table_name='user', rm_id_random=True): pass
else:
return False
log.setLevel(logging.DEBUG)
log.debug(user_obj_up_result)
log.debug(f'Returning the existing user_id: {user_id}')
else:
if user_obj_in_result := sql_insert(data=user_obj_data, table_name='user', rm_id_random=True, id_random_length=8): pass
else:
return False
log.setLevel(logging.DEBUG)
log.debug(user_obj_in_result)
user_id = user_obj_in_result
log.debug(f'Returning the new user_id: {user_id}')
return user_id
# ### END ### API User Methods ### create_user_obj() ###
# ### BEGIN ### API User Methods ### load_user_obj() ###
def load_user_obj(user_id:int|str, inc_roles:bool=False, inc_contact:bool=False, inc_organization:bool=False, inc_person:bool=False) -> User_Base:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL

35
app/models/user_model.py
View File

@@ -29,11 +29,17 @@ class User_New_Base(BaseModel):
)
account_id_random: str
account_id: Optional[int]
contact_id_random: Optional[str]
contact_id: Optional[int]
organization_id_random: Optional[str]
organization_id: Optional[int]
person_id_random: Optional[str]
person_id: Optional[int]
username: str
name: str
email: str
new_password: str
new_password: str = Field(default_factory=lambda:secrets.token_urlsafe(default_num_bytes))
password: Optional[str]
allow_auth_key: Optional[int]
@@ -82,6 +88,33 @@ class User_New_Base(BaseModel):
return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account')
return None
@validator('contact_id', always=True)
def contact_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['contact_id_random']:
return redis_lookup_id_random(record_id_random=values['contact_id_random'], table_name='contact')
return None
@validator('organization_id', always=True)
def organization_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['organization_id_random']:
return redis_lookup_id_random(record_id_random=values['organization_id_random'], table_name='organization')
return None
@validator('person_id', always=True)
def person_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['person_id_random']:
return redis_lookup_id_random(record_id_random=values['person_id_random'], table_name='person')
return None
@validator('password', always=True)
def hash_new_password(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)