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

Save before hosted_file_link DB field name changes.

Browse Source
This commit is contained in:
Scott Idem
2021-06-18 09:42:29 -04:00
parent 47ce2380be
commit 6d4463e57e
4 changed files with 183 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
app/methods/account_methods.py
View File

@@ -175,17 +175,6 @@ def load_account_obj(
account_obj.archive_list = archive_dict_list
else: account_obj.archive_list = []
# if inc_contact_list:
# if contact_dict_list := load_contact_obj_list(
# account_id = account_id,
# limit = limit,
# model_as_dict = model_as_dict,
# enabled = enabled,
# inc_address = inc_address,
# ):
# account_obj.contact_list = contact_dict_list
# else: account_obj.contact_list = []
# Updated 2021-06-17
if inc_contact_list:
if contact_rec_list_result := get_contact_rec_list(
@@ -247,6 +236,28 @@ def load_account_obj(
account_obj.event_list = event_dict_list
else: account_obj.event_list = []
# Updated 2021-06-17
if inc_hosted_file_list:
if hosted_file_rec_list_result := get_hosted_file_rec_list(
for_obj_type = 'account',
for_obj_id = account_id,
limit = limit,
enabled = enabled,
):
hosted_file_dict_list = []
for hosted_file_rec in hosted_file_rec_list_result:
hosted_file_dict_list.append(
load_hosted_file_obj(
hosted_file_id = hosted_file_rec.get('hosted_file_id', None),
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_hosted_file_link_list = inc_hosted_file_line_list,
)
)
account_obj.hosted_file_list = hosted_file_dict_list
else: account_obj.hosted_file_list = []
# Updated 2021-06-17
if inc_order_list:
if order_rec_list_result := get_order_rec_list(

70
app/methods/hosted_file_methods.py
View File

@@ -38,7 +38,7 @@ def load_hosted_file_obj(
limit: int = 1000,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
# inc_x: bool = False,
inc_hosted_file_link_list: bool = False,
) -> Hosted_File_Base|dict|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -213,9 +213,8 @@ async def save_file(
def create_hosted_file_link(
account_id: int|str,
hosted_file_id: int|str,
for_object_type: str,
for_object_id: int|str,
# for_object_id_random: str,
link_to_obj_type: str,
link_to_obj_id: int|str,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -224,15 +223,14 @@ def create_hosted_file_link(
else: return False
if hosted_file_id := redis_lookup_id_random(record_id_random=hosted_file_id, table_name='hosted_file'): pass
else: return False
if for_object_id := redis_lookup_id_random(record_id_random=for_object_id, table_name=for_object_type): pass
if link_to_obj_id := redis_lookup_id_random(record_id_random=link_to_obj_id, table_name=link_to_obj_type): pass
else: return False
hosted_file_link_data: dict = {}
hosted_file_link_data['account_id'] = account_id
hosted_file_link_data['hosted_file_id'] = hosted_file_id
hosted_file_link_data['object_type'] = for_object_type
hosted_file_link_data['object_id'] = for_object_id
hosted_file_link_data['link_to_type'] = link_to_obj_type # Should this be renamed to "link_to_obj_type" for clarity?
hosted_file_link_data['link_to_id'] = link_to_obj_id # Should this be renamed to "link_to_obj_id" for clarity?
# NOTE: Currently sql_insert does not handel all successful inserts correctly. If there is not an autonum ID then it will return 0 as the ID.
if hosted_file_link_data_in_result := sql_insert(data=hosted_file_link_data, table_name='hosted_file_link', id_random_length=0): pass # This should be improved
@@ -247,3 +245,59 @@ def create_hosted_file_link(
return True
# ### END ### API Hosted File Route ### hosted_file_link() ###
# ### BEGIN ### API Hosted File Methods ### get_hosted_file_rec_list() ###
# This needs to be improved. Currently it does not really do anything.
# Need to allow for list by account? Probably have the same actual hosted file have two hosted_file entries if it was uploaded for two separate accounts.
def get_hosted_file_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if for_obj_id := redis_lookup_id_random(record_id_random=for_obj_id, table_name='for_obj_type'): pass
else: return False
data = {}
data[f'{for_obj_type}_id'] = for_obj_id
# data['for_obj_type'] = for_obj_type
sql_obj_type_id = f'`tbl`.{for_obj_type}_id = :{for_obj_type}_id'
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
data['enable'] = True
sql_enabled = f'AND `tbl`.enable = :enable'
elif enabled == 'disabled':
data['enable'] = False
sql_enabled = f'AND `tbl`.enable = :enable'
elif enabled == 'all':
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT `tbl`.id AS 'hosted_file_id', `tbl`.id_random AS 'hosted_file_id_random'
FROM `hosted_file` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if hosted_file_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
hosted_file_rec_li = hosted_file_rec_li_result
else:
hosted_file_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(hosted_file_rec_li_result)
return hosted_file_rec_li
# ### END ### API Hosted File Methods ### get_hosted_file_rec_list() ###

85
app/models/hosted_file_link_models.py Normal file
View File

@@ -0,0 +1,85 @@
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 app.db_sql import redis_lookup_id_random
from app.lib_general import log, logging
from .common_field_schema import base_fields, default_num_bytes
class Hosted_File_Link_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['hosted_file_link_id_random'],
alias='hosted_file_link_id_random',
default_factory=lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
#alias='hosted_file_link_id'
)
account_id_random: Optional[str]
account_id: Optional[int]
hosted_file_id_random: Optional[str]
hosted_file_id: Optional[int]
link_to_type: Optional[str] # Should this be renamed to "link_to_obj_type" for clarity?
link_to_id_random: Optional[str] # Should this be renamed to "link_to_obj_id_random" for clarity?
link_to_id: Optional[int] # Should this be renamed to "link_to_obj_id" for clarity?
# notes: Optional[str]
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
# Including other related objects
link_to: Optional[dict] # Should this be renamed to "link_to_obj" or leave off the "_obj"?
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('hosted_file_link_id_random', always=True)
def hosted_file_link_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 hosted_file_link_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='hosted_file_link')
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('object_id', always=True)
def object_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['object_id_random'] and values['object_type']:
return redis_lookup_id_random(record_id_random=values['object_id_random'], table_name=values['object_type'])
return None
class Config:
underscore_attrs_are_private = True
fields = base_fields
Hosted_File_Base.update_forward_refs()

16
app/models/hosted_file_models.py
View File

@@ -22,8 +22,8 @@ class Hosted_File_Base(BaseModel):
id: Optional[int] = Field(
#alias='hosted_file_id'
)
#account_id_random: Optional[str]
#account_id: Optional[int]
account_id_random: Optional[str]
account_id: Optional[int]
hash_sha256: Optional[str]
title: Optional[str]
@@ -59,6 +59,9 @@ class Hosted_File_Base(BaseModel):
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
# Including other related objects
hosted_file_link_list: Optional[list] # Hosted_File_Base()
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('hosted_file_id_random', always=True)
@@ -80,6 +83,15 @@ class Hosted_File_Base(BaseModel):
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='hosted_file')
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
class Config:
underscore_attrs_are_private = True
fields = base_fields