Finally updating this...

This commit is contained in:
Scott Idem
2021-03-05 17:27:16 -05:00
parent 420180f20c
commit 28cf7ecf11
14 changed files with 1235 additions and 108 deletions

View File

@@ -0,0 +1,119 @@
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 ..lib_general import *
from ..log import *
from .common_field_schema import base_fields, default_num_bytes
#from .account_model import Account_Base
class Address_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
#from .account_model import Account_Base
id_random: Optional[str] = Field(
**base_fields['address_id_random'],
alias='address_id_random',
default_factory=lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
#alias='address_id'
)
account_id_random: Optional[str]
account_id: Optional[int]
for_type: Optional[str]
for_id_random: Optional[str]
for_id: Optional[int] #organization: Optional[Organization_Base] = Organization_Base()
name: Optional[str]
attention_to: Optional[str]
organization_name: Optional[str]
line_1: Optional[str]
line_2: Optional[str]
line_3: Optional[str]
city: Optional[str]
country_subdivision_code: Optional[str]
state_province: Optional[str]
postal_code: Optional[str]
country_alpha_2_code: Optional[str]
country: Optional[str]
lu_time_zone_id: Optional[str]
timezone: Optional[str]
latitude: Optional[str]
longitude: Optional[str]
map_url: Optional[str]
congressional_district: Optional[str]
#priority: Optional[int]
#sort: Optional[int]
#group: Optional[str]
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
#account: Optional[Account_Base] = Account_Base()
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('address_id_random', always=True)
def address_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 address_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='address')
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('organization_id', always=True)
#def organization_id_lookup(cls, v, values, **kwargs):
#log.setLevel(logging.WARNING)
#log.debug(locals())
#if values['organization_id']:
#return redis_lookup_id_random(record_id_random=values['organization_id'], table_name='organization')
#return None
@validator('for_id', always=True)
def for_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['for_id_random'] and values['for_type']:
return redis_lookup_id_random(record_id_random=values['for_id_random'], table_name=values['for_type'])
return None
class Config:
underscore_attrs_are_private = True
fields = base_fields
Address_Base.update_forward_refs()

80
app/routers/api_crud.py Normal file
View File

@@ -0,0 +1,80 @@
import datetime
#from datetime import datetime, time, timedelta
from fastapi import APIRouter, Depends, Header, HTTPException, Query, 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 .journal_models import *
#from .user_models import *
from .user_model import *
from .response_model import *
router = APIRouter()
# Working on the basic API CRUD - STI 2021-03-05
#@router.get('/{object_l1}/list')
@router.get('/{object_l1}/{object_id}/list')
@router.get('/{object_l1}/{object_l2}/{object_id}/list')
@router.get('/{object_l1}/{object_l2}/{object_id}/{object_l3}/list')
async def get_obj_li(object_l1: str=None, object_l2: str=None, object_l3: str=None, object_id: str=None, x_account_id: str = Header(...)):
response_data = {}
response_data['object_l1'] = object_l1
response_data['object_l2'] = object_l2
response_data['object_l3'] = object_l3
response_data['object_id'] = object_id
response_data['list'] = 'li'
sql_result = sql_select(table_name='user', record_id=1)
response_data['sql_result'] = sql_result
return response_data
@router.get('/{object_l1}/{object_id}')
@router.get('/{object_l1}/{object_l2}/{object_id}')
@router.get('/{object_l1}/{object_l2}/{object_l3}/{object_id}')
async def get_obj(object_l1: str=None, object_l2: str=None, object_l3: str=None, object_id: str=None, x_account_id: str = Header(...),
qry_str: Optional[str] = Query(None, max_length=50),
qry_int: Optional[int] = None,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.debug(by_alias)
log.debug(exclude_unset)
log.debug(qry_str)
log.debug(qry_int)
response_data = {}
response_data['object_l1'] = object_l1
response_data['object_l2'] = object_l2
response_data['object_l3'] = object_l3
response_data['object_id'] = object_id
data = {}
data['id_random'] = 1
sql_select_str = f"""
SELECT `user`.id AS 'user_id', `user`.id_random AS 'user_id_random', username, name, email, super
FROM `user` AS `user`
WHERE `user`.id = :id_random;
"""
#sql_result = sql_select(table_name='user', record_id=1)
sql_result = sql_select(sql=sql_select_str, data=data)
resp_data = User_Base(**sql_result).dict(by_alias=by_alias, exclude_unset=exclude_unset)
#response_data['sql_result'] = sql_result
resp = mk_resp(data=resp_data)
return resp

View File

@@ -0,0 +1,69 @@
import copy, datetime, hashlib, logging, os, pytz, redis, secrets
default_num_bytes = 8 # URL safe 8 bytes is 11 characters long and 16 bytes is 22 characters long
xxx_id_random_field_schema: dict = {
'title': 'XXX ID Random',
'description': 'This is an id_random field for this object.',
'min_length': 11,
'max_length': 22,
'example': secrets.token_urlsafe(8) # random each reloading of app with: secrets.token_urlsafe(8)
}
xxx_id_random_field_schema_default: dict = copy.copy(xxx_id_random_field_schema)
xxx_id_random_field_schema_default['default_factory'] = lambda:secrets.token_urlsafe(8)
created_updated_on_field_schema: dict = {
'title': 'Created or Updated On',
'description': 'This is the created or updated on timestamp field for this object. It is filled in by the SQL DB.',
'example': '2021-12-31T21:10:10'
}
base_fields = {}
#base_fields['id_random'] = xxx_id_random_field_schema_default
base_fields['obj_id_random'] = xxx_id_random_field_schema # General or generic object_id_random
base_fields['account_id_random'] = xxx_id_random_field_schema
base_fields['address_id_random'] = xxx_id_random_field_schema
base_fields['archive_id_random'] = xxx_id_random_field_schema
base_fields['contact_id_random'] = xxx_id_random_field_schema
base_fields['event_exhibit_id_random'] = xxx_id_random_field_schema
base_fields['event_file_id_random'] = xxx_id_random_field_schema
base_fields['event_id_random'] = xxx_id_random_field_schema
base_fields['event_presentation_id_random'] = xxx_id_random_field_schema
base_fields['event_registration_id_random'] = xxx_id_random_field_schema
base_fields['fundraising_id_random'] = xxx_id_random_field_schema
base_fields['hosted_file_id_random'] = xxx_id_random_field_schema
base_fields['membership_id_random'] = xxx_id_random_field_schema
base_fields['membership_profile_id_random'] = xxx_id_random_field_schema
base_fields['order_cart_id_random'] = xxx_id_random_field_schema
base_fields['order_cart_line_id_random'] = xxx_id_random_field_schema
base_fields['order_id_random'] = xxx_id_random_field_schema
base_fields['order_line_id_random'] = xxx_id_random_field_schema
base_fields['order_transaction_id_random'] = xxx_id_random_field_schema
base_fields['organization_id_random'] = xxx_id_random_field_schema
base_fields['page_id_random'] = xxx_id_random_field_schema
base_fields['person_id_random'] = xxx_id_random_field_schema
base_fields['post_id_random'] = xxx_id_random_field_schema
base_fields['post_comment_id_random'] = xxx_id_random_field_schema
base_fields['product_id_random'] = xxx_id_random_field_schema
base_fields['site_id_random'] = xxx_id_random_field_schema
base_fields['site_domain_id_random'] = xxx_id_random_field_schema
base_fields['user_id_random'] = xxx_id_random_field_schema
base_fields['created_on'] = created_updated_on_field_schema
base_fields['updated_on'] = created_updated_on_field_schema
base_fields['obj_type'] = {}
base_fields['obj_id_random'] = xxx_id_random_field_schema_default
base_fields['obj_id_rand'] = xxx_id_random_field_schema_default
base_fields['obj_id'] = {}
base_fields['obj_name'] = {}
base_fields['obj_notes'] = {}
base_fields['for_id_random'] = xxx_id_random_field_schema_default
#xxx_id_random_field_schema['alias'] = 'order_id_random'
#base_fields['id_random'] = xxx_id_random_field_schema_default
#xxx_id_random_field_schema['alias'] = 'user_id_random_x'
#c = {'alias': 'user_id_random_x'}
#base_fields['user_id_random'] = combine_dict(xxx_id_random_field_schema, c)

View File

@@ -0,0 +1,129 @@
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 ..lib_general import *
from ..log import *
from .common_field_schema import base_fields, default_num_bytes
#from .account_model import Account_Base
from .address_model import Address_Base
class Contact_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
#from .account_model import Account_Base
#from .address_model import Address_Base
id_random: Optional[str] = Field(
**base_fields['contact_id_random'],
alias='contact_id_random',
default_factory=lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
#alias='contact_id'
)
account_id_random: Optional[str]
account_id: Optional[int]
address_id_random: Optional[str]
address_id: Optional[int]
for_type: Optional[str]
for_id_random: Optional[str]
for_id: Optional[int]
name: Optional[str]
title: Optional[str]
tagline: Optional[str]
description: Optional[str]
lu_time_zone_id: Optional[str]
timezone: Optional[str]
email: Optional[str]
website: Optional[str]
website_name: Optional[str]
phone_mobile: Optional[str]
phone_home: Optional[str]
phone_office: Optional[str]
phone_land: Optional[str]
phone_fax: Optional[str]
facebook: Optional[str]
instagram: Optional[str]
twitter: Optional[str]
linkedin: Optional[str]
other_site_url: Optional[str]
other_site_name: Optional[str]
other_text: Optional[str]
other_json: Optional[Json]
priority: Optional[int]
sort: Optional[int]
group: Optional[str]
#account: Optional[Account_Base] = Account_Base()
address: Optional[Address_Base] = Address_Base()
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('contact_id_random', always=True)
def contact_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 contact_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='contact')
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('address_id', always=True)
def address_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values.get('address_id_random', None):
return redis_lookup_id_random(record_id_random=values['address_id_random'], table_name='address')
return None
@validator('for_id', always=True)
def for_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['for_id_random'] and values['for_type']:
return redis_lookup_id_random(record_id_random=values['for_id_random'], table_name=values['for_type'])
return None
class Config:
underscore_attrs_are_private = True
fields = base_fields
Contact_Base.update_forward_refs()

View File

@@ -0,0 +1,55 @@
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, PrivateAttr, ValidationError, validator
from .common_field_schema import base_fields
class Core_Object_Base(BaseModel):
app.logger.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
app.logger.debug(locals())
obj_type: str
obj_id_random: str # alias this one based on obj_type?
obj_id_rand: str # alias this one based on obj_type?
obj_id: int # alias this one?
obj_name: Optional[str]
id_random: str # alias this one?
id: int # alias this one?
account_id_random: Optional[str]
account_id: 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)
class Example_Object_Base(Core_Object_Base): # Based on Core_Object_Base
title: Optional[str] = None
description: Optional[str] = None
password_set_on: Optional[datetime.datetime] = None
archive_on: Optional[datetime.datetime] = None
logged_in_on: Optional[datetime.datetime] = None
last_activity_on: Optional[datetime.datetime] = None
other_random_fields: dict
list_of_: Optional[dict] = {}
# Create, Read/Get, Update, Delete
# CRUD or CGUD
# def create_object(object_data):
# return False # True, False, or None or object_data
# def get_object(object_id):
# return object_data # False or None
# def update_object(object_id, object_data):
# return False # True, False, or None or object_data
# def delete_object(object_id):
# return False # True, False, or None or object_data

View File

@@ -0,0 +1,62 @@
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, PrivateAttr, ValidationError, validator
from ..lib_general import *
from ..log import *
from .common_field_schema import base_fields
from app.config import settings
# The pydantic BaseModel to help make consistent REST responses - STI 2021-03-05
class Resp_Body_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
data: Union[dict, list]
meta: Optional[dict]
# The make response function for REST - STI 2021-03-05
def mk_resp(data={}, dict_to_json=None, status_code=200, status_message=None, status_name=None, success=True, details=None, by_alias=True, exclude_unset=True):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if data is None: data = { 'result': None }
elif data == False: data = { 'result': False }
elif data == True: data = { 'result': True }
resp_body = {}
resp_body['data'] = data
resp_body['meta'] = {}
resp_body['meta']['details'] = details
resp_body['meta']['status_code'] = status_code
resp_body['meta']['status_message'] = settings.HTTP_STATUS_LI[status_code]['message']
resp_body['meta']['status_name'] = settings.HTTP_STATUS_LI[status_code]['name']
resp_body['meta']['success'] = success
if isinstance(data, bool):
resp_body['meta']['data_type'] = 'bool'
elif isinstance(data, int):
resp_body['meta']['data_type'] = 'int'
elif isinstance(data, str):
resp_body['meta']['data_type'] = 'str'
elif isinstance(data, dict):
resp_body['meta']['data_type'] = 'dict'
elif isinstance(data, list):
resp_body['meta']['data_type'] = 'list'
resp_body['meta']['data_list_count'] = len(data)
log.debug(type(resp_body['data']))
resp_body = Resp_Body_Base(**resp_body).dict(by_alias=by_alias, exclude_unset=exclude_unset)
#resp_body_json = resp_body.json(by_alias=True, exclude_unset=False)
#response = app.response_class(
#response=resp_body_json,
#status=status_code,
#mimetype='application/json'
#)
return resp_body

140
app/routers/user_model.py Normal file
View File

@@ -0,0 +1,140 @@
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 ..lib_general import *
from ..log import *
from .common_field_schema import base_fields, default_num_bytes
#from .account_model import Account_Base
from .contact_model import Contact_Base
#from .organization_model import Organization_Base
#from .person_model import Person_Base
class User_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
#from .account_model import Account_Base
#from .contact_model import Contact_Base
#from .organization_model import Organization_Base
#from .person_model import Person_Base
#if TYPE_CHECKING:
#from .person_model import Person_Base
id_random: Optional[str] = Field(
**base_fields['user_id_random'],
alias='user_id_random',
default_factory=lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
#alias='user_id'
)
account_id_random: Optional[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: Optional[str]
name: Optional[str]
email: Optional[str]
email_verified: Optional[bool]
password: Optional[str]
auth_key: Optional[str]
enable: Optional[bool]
enable_from: Optional[datetime.datetime] = None
enable_to: Optional[datetime.datetime] = None
super: Optional[bool]
manager: Optional[bool]
administrator: Optional[bool]
public: Optional[bool]
verified: Optional[bool]
status_id: Optional[int]
status_name: Optional[str]
password_set_on: Optional[datetime.datetime] = None
password_reset_token: Optional[str] = None
password_reset_expire_on: Optional[datetime.datetime] = None
logged_in_on: Optional[datetime.datetime] = None
last_activity_on: Optional[datetime.datetime] = None
#account: Optional[Account_Base]# = Account_Base()
contact: Optional[Contact_Base]# = Contact_Base()
#organization: Optional[Organization_Base]# = Organization_Base()
#person: Optional[Person_Base]# = Person_Base()
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('user_id_random', always=True)
def user_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 user_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='user')
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('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
class Config:
underscore_attrs_are_private = True
fields = base_fields
User_Base.update_forward_refs()