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

refactor(routers): comment out legacy endpoints across multiple routers

Browse Source 
Disabled legacy routes that are superseded by V3 equivalents. Code is
commented out (not deleted) pending final verification and cleanup pass.

- registry.py: remove sql, lookup (/lu), websockets, websockets_redis;
  clean up dead imports (contact, event_person, etc.)
- data_store.py: comment out legacy CRUD and code-lookup endpoints;
  keep V3 code-lookup routes active; add TODO for action path rename
- api.py: comment out Api_Base CRUD, get_id (internal ID leak),
  and sql_test (debug) endpoints
- aether_cfg.py: comment out legacy Flask cfg endpoint
- user.py: comment out legacy user endpoints
- util_email.py: minor cleanup

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Scott Idem
2026-03-24 19:22:45 -04:00
parent c7f1341b1e
commit 8eb699efe5
6 changed files with 501 additions and 472 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
app/routers/api.py
View File

@@ -62,13 +62,13 @@ async def authenticate_passcode(
if matched_role:
log.info(f"Auth Success: Verified '{matched_role}' passcode for site {site_id}")
# 4. Resolve Account Context
account_id_random = record.get('account_id_random')
if not account_id_random:
if account_id_int := record.get('account_id'):
account_id_random = get_id_random(record_id=account_id_int, table_name='account')
# 5. Mint JWT
payload = {
'account_id': account_id_random,
@@ -81,13 +81,13 @@ async def authenticate_passcode(
'role': matched_role
})
}
token = sign_jwt(
secret_key=settings.JWT_KEY,
ttl=3600 * 24, # 24 hour session
**payload
)
return mk_resp(data={'jwt': token, 'account_id': account_id_random, 'role': matched_role}, response=response)
else:
log.warning(f"Auth Failed: Invalid passcode for site {site_id}")
@@ -167,6 +167,8 @@ async def get_api_temp_token(
# --- Jitsi Token ---
# NOTE: This is still actively used by IDAA for their video conferences using self hosted Jitsi. Thi is actually live. We do need to change the app secret once things have stabilized.
JWT_APP_ID = "my_jitsi_app_id"
JWT_APP_SECRET = "my_jitsi_app_secret-9876543210"
JITSI_DOMAIN = "jitsi.dgrzone.com"
@@ -211,40 +213,43 @@ async def create_jitsi_jwt(request_data: JitsiTokenRequest = Body(...)):
raise HTTPException(status_code=500, detail=f"Failed to create JWT: {str(e)}")
# --- Api_Base CRUD ---
# LEGACY (disabled) - superseded by V3 CRUD: /v3/crud/api/
@router.post('', response_model=Resp_Body_Base)
async def post_api_obj(obj: Api_Base, x_account_id: str = Header(...)):
return post_obj_template(obj_type='api', data=obj.dict(by_alias=False, exclude_unset=True), return_obj=True)
# @router.post('', response_model=Resp_Body_Base)
# async def post_api_obj(obj: Api_Base, x_account_id: str = Header(...)):
# return post_obj_template(obj_type='api', data=obj.dict(by_alias=False, exclude_unset=True), return_obj=True)
@router.patch('/{obj_id}', response_model=Resp_Body_Base)
async def patch_api_obj(obj_id: str, obj: Api_Base, x_account_id: str = Header(...)):
data = obj.dict(by_alias=False, exclude_unset=True)
data['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name='api')
return patch_obj_template(obj_type='api', data=data, obj_id=obj_id, return_obj=True)
# @router.patch('/{obj_id}', response_model=Resp_Body_Base)
# async def patch_api_obj(obj_id: str, obj: Api_Base, x_account_id: str = Header(...)):
# data = obj.dict(by_alias=False, exclude_unset=True)
# data['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name='api')
# return patch_obj_template(obj_type='api', data=data, obj_id=obj_id, return_obj=True)
@router.get('/list', response_model=Resp_Body_Base)
async def get_api_obj_li(for_obj_type: Optional[str] = Query(None), for_obj_id: Optional[str] = Query(None), x_account_id: str = Header(...)):
return get_obj_li_template(obj_type='api', for_obj_type=for_obj_type, for_obj_id=for_obj_id)
# @router.get('/list', response_model=Resp_Body_Base)
# async def get_api_obj_li(for_obj_type: Optional[str] = Query(None), for_obj_id: Optional[str] = Query(None), x_account_id: str = Header(...)):
# return get_obj_li_template(obj_type='api', for_obj_type=for_obj_type, for_obj_id=for_obj_id)
@router.get('/{obj_id}', response_model=Resp_Body_Base)
async def get_api_obj(obj_id: str, x_account_id: str = Header(...)):
return get_obj_template(obj_type='api', obj_id=obj_id)
# @router.get('/{obj_id}', response_model=Resp_Body_Base)
# async def get_api_obj(obj_id: str, x_account_id: str = Header(...)):
# return get_obj_template(obj_type='api', obj_id=obj_id)
@router.delete('/{obj_id}', response_model=Resp_Body_Base)
async def delete_api_obj(obj_id: str, x_account_id: str = Header(...)):
return delete_obj_template(obj_type='api', obj_id=obj_id)
# @router.delete('/{obj_id}', response_model=Resp_Body_Base)
# async def delete_api_obj(obj_id: str, x_account_id: str = Header(...)):
# return delete_obj_template(obj_type='api', obj_id=obj_id)
@router.get('/get_id/{object_type}/{object_id_random}', response_model=Resp_Body_Base)
async def get_api_object_id(object_type: str, object_id_random: str):
if object_id := redis_lookup_id_random(record_id_random=object_id_random, table_name=object_type):
return mk_resp(data={ 'object_id': object_id})
return mk_resp(data=None, status_code=404)
# LEGACY (disabled) - exposes internal integer IDs, breaks id_random abstraction
# @router.get('/get_id/{object_type}/{object_id_random}', response_model=Resp_Body_Base)
# async def get_api_object_id(object_type: str, object_id_random: str):
# if object_id := redis_lookup_id_random(record_id_random=object_id_random, table_name=object_type):
# return mk_resp(data={ 'object_id': object_id})
# return mk_resp(data=None, status_code=404)
@router.get('/sql_test', tags=['Testing'])
async def sql_test(response: Response = Response):
sql = text("SELECT NOW() as current_time, VERSION() as version")
try:
result = db.execute(sql).fetchone()
return mk_resp(data={"current_time": str(result[0]), "version": result[1]})
except Exception as e:
return mk_resp(data=False, status_code=500, details=str(e), response=response)
# LEGACY (disabled) - testing/debug endpoint
# @router.get('/sql_test', tags=['Testing'])
# async def sql_test(response: Response = Response):
# sql = text("SELECT NOW() as current_time, VERSION() as version")
# try:
# result = db.execute(sql).fetchone()
# return mk_resp(data={"current_time": str(result[0]), "version": result[1]})
# except Exception as e:
# return mk_resp(data=False, status_code=500, details=str(e), response=response)