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

feat(data_store): implement V3 cascading lookup and ID Vision standardization

Browse Source 
- Added GET /v3/data_store/code/{code} with hierarchical context-aware fallback.
- Implemented ID Vision standard in Data_Store_Base (string IDs, internal int exclusion).
- Enhanced Data_Store_Base robustness to handle stringified 'NULL' values from the database.
- Fixed legacy router bugs by removing undefined parameters (inc_event_cfg, inc_event_location).
- Corrected type hints and resolved UnboundLocalError in data_store methods.
- Updated Frontend Integration Guide with Section 8: Data Store V3.
- Added unified E2E test script: tests/e2e/test_e2e_v3_data_store_lookup.py.
This commit is contained in:
Scott Idem
2026-01-28 16:51:48 -05:00
parent 9c0aae9a6d
commit fdcc859017
5 changed files with 224 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
app/methods/data_store_methods.py
View File

@@ -55,7 +55,7 @@ def load_data_store_obj(
def load_data_store_obj_w_code(
account_id: int,
code: str,
for_type: int = None,
for_type: str = None,
for_id: int = None,
enabled: str = 'enabled', # enabled, disabled, all
limit: int = 1,
@@ -123,11 +123,11 @@ def load_data_store_obj_w_code(
try:
data_store_obj = Data_Store_Base(**data_store_rec)
data_store_obj_li.append(data_store_obj)
log.debug(data_store_obj)
except ValidationError as e:
log.error(e.json())
data_store_obj_li.append(None)
# return False
log.debug(data_store_obj)
else: pass
log.info(f'Found {len(data_store_obj_li)} Data Store records with code: {code} for Account ID: {account_id} and For Type: {for_type} and For ID: {for_id}')

124
app/models/data_store_models.py
View File

@@ -1,7 +1,7 @@
import datetime, pytz
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator
from typing import Dict, List, Optional, Set, Union, ClassVar
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator, root_validator
from app.db_sql import redis_lookup_id_random
from app.lib_general import log, logging
@@ -11,39 +11,37 @@ from app.models.common_field_schema import base_fields
# ### BEGIN ### API Data Store Models ### Data_Store_Base() ###
class Data_Store_Base(BaseModel):
# log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['data_store_id_random'],
alias = 'data_store_id_random',
)
id: Optional[int] = Field(
alias = 'data_store_id'
)
# --- Standardized Vision IDs (Strings) ---
id: Optional[str] = Field(None, **base_fields['data_store_id_random'])
data_store_id: Optional[str] = Field(None, **base_fields['data_store_id_random'])
account_id: Optional[str] = Field(None, **base_fields['account_id_random'])
person_id: Optional[str] = Field(None, **base_fields['person_id_random'])
user_id: Optional[str] = Field(None, **base_fields['user_id_random'])
account_id_random: Optional[str]
account_id: Optional[int]
# Internal Integer IDs (Excluded from API)
# We use Optional[Union[int, str]] here to prevent validation crashes
# if the DB returns stringified integers or "NULL" strings.
id_int: Optional[Union[int, str]] = Field(None, alias='id', exclude=True)
account_id_int: Optional[Union[int, str]] = Field(None, alias='account_id', exclude=True)
person_id_int: Optional[Union[int, str]] = Field(None, alias='person_id', exclude=True)
user_id_int: Optional[Union[int, str]] = Field(None, alias='user_id', exclude=True)
for_type: Optional[str]
for_id_random: Optional[str]
for_id: Optional[int]
person_id_random: Optional[str]
person_id: Optional[int]
user_id_random: Optional[str]
user_id: Optional[int]
for_id: Optional[str] # Random ID string
for_id_random: Optional[str] # Svelte often uses this name
for_id_int: Optional[Union[int, str]] = Field(None, alias='for_id', exclude=True)
code: Optional[str]
name: Optional[str]
description: Optional[str]
type: Optional[str] # html, json, md, text
# The JSON fields are case sensitive
# json: Optional[str] # "json" is reserved; need to change field name? json_str?
json_str: Optional[Union[Json, None]] = Field(
alias = 'json',
)
@@ -71,58 +69,46 @@ class Data_Store_Base(BaseModel):
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
# Including convenience data
# This is only for convenience. Probably going to keep unless it causes a problem.
# Including JSON data
# other_json: Optional[Json]
# meta_json: Optional[Json]
# Including other related objects
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
@validator('id', always=True)
def data_store_id_lookup(cls, v, values, **kwargs):
if isinstance(v, int) and v > 0: return v
elif id_random := values.get('id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='data_store')
return None
@root_validator(pre=True)
def map_v3_ids(cls, values):
"""
Vision Transformer:
Map DB keys to clean API keys and strip internal integers.
"""
# 0. Scrub stringified NULLs from database
for k, v in list(values.items()):
if isinstance(v, str) and v.upper() == 'NULL':
values[k] = None
@validator('account_id', always=True)
def account_id_lookup(cls, v, values, **kwargs):
if isinstance(v, int) and v > 0: return v
elif id_random := values.get('account_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='account')
return None
@validator('for_id', always=True)
def for_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if isinstance(v, int) and v > 0: return v
elif values.get('for_id_random') and values.get('for_type'):
for_id_random = values.get('for_id_random')
for_type = values.get('for_type')
return redis_lookup_id_random(record_id_random=for_id_random, table_name=for_type)
return None
@validator('person_id', always=True)
def person_id_lookup(cls, v, values, **kwargs):
if isinstance(v, int) and v > 0: return v
elif id_random := values.get('person_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='person')
return None
@validator('user_id', always=True)
def user_id_lookup(cls, v, values, **kwargs):
if isinstance(v, int) and v > 0: return v
elif id_random := values.get('user_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='user')
return None
# 1. Map Random Strings to Clean Names
if rid := values.get('id_random') or values.get('data_store_id_random'):
values['id'] = rid
values['data_store_id'] = rid
if a_rid := values.get('account_id_random'):
values['account_id'] = a_rid
if p_rid := values.get('person_id_random'):
values['person_id'] = p_rid
if u_rid := values.get('user_id_random'):
values['user_id'] = u_rid
if f_rid := values.get('for_id_random'):
values['for_id'] = f_rid
values['for_id_random'] = f_rid
# 2. Prevent "Collision Population"
# We only want strings in our primary ID fields.
# If the key exists and isn't a string, it's a DB integer; remove it
# so it doesn't fail length validation on the string fields.
for k in ['id', 'data_store_id', 'account_id', 'person_id', 'user_id', 'for_id']:
if k in values and not isinstance(values[k], str):
del values[k]
return values
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
allow_population_by_field_name = False
fields = base_fields
# ### END ### API Data Store Models ### Data_Store_Base() ###

48
app/routers/data_store.py
View File

@@ -20,7 +20,7 @@ router = APIRouter()
# ### BEGIN ### API Data Store Routers ### post_data_store_obj() ###
# Updated 2022-03-11
# Updated 2026-01-28
@router.post('/data_store', response_model=Resp_Body_Base)
async def post_data_store_obj(
data_store_obj: Data_Store_Base,
@@ -47,8 +47,6 @@ async def post_data_store_obj(
if return_obj:
data_store_obj = load_data_store_obj(
data_store_id = data_store_id,
inc_event_cfg = inc_event_cfg,
inc_event_location = inc_event_location,
)
data = data_store_obj
else:
@@ -103,7 +101,7 @@ async def patch_data_store_obj(
# ### BEGIN ### API Data Store ### get_data_store_obj() ###
# Updated 2022-03-11
# Updated 2026-01-28
@router.get('/data_store/{data_store_id}', response_model=Resp_Body_Base)
async def get_data_store_obj(
data_store_id: str = Path(min_length=11, max_length=22),
@@ -121,8 +119,6 @@ async def get_data_store_obj(
data_store_id = data_store_id,
limit = commons.limit,
enabled = commons.enabled,
inc_event_cfg = inc_event_cfg,
inc_event_location = inc_event_location,
):
log.info('Loading successful. Returning result')
return mk_resp(data=data_store_rec_result, response=commons.response)
@@ -135,6 +131,46 @@ async def get_data_store_obj(
# ### END ### API Data Store ### get_data_store_obj() ###
# ### BEGIN ### API Data Store ### get_v3_data_store_obj_w_code() ###
# NEW V3 Endpoint for Code Lookup
# Updated 2026-01-28
from app.lib_general_v3 import AccountContext, get_account_context, SerializationParams, StatusFilterParams
@router.get('/v3/data_store/code/{data_store_code}', response_model=Resp_Body_Base, tags=['Data Store V3'])
async def get_v3_data_store_obj_w_code(
data_store_code: str = Path(min_length=3, max_length=50),
for_type: Optional[str] = Query(None, min_length=1, max_length=25),
for_id: Optional[str] = Query(None, min_length=11, max_length=22),
account: AccountContext = Depends(get_account_context),
serialization: SerializationParams = Depends(),
status_filter: StatusFilterParams = Depends(),
):
"""
V3 Standardized Data Store Lookup.
Uses JWT-based AccountContext and supports cascading fallback logic (Object > Account > Global).
"""
log.setLevel(logging.INFO)
# Map V3 params to the shared handler
# We create a dummy Common_Route_Params object to satisfy the handler's interface
# while using the more secure V3 dependencies.
v3_commons = Common_Route_Params(
x_account_id=account.account_id,
x_account_id_random=account.account_id_random,
enabled=status_filter.enabled,
response=Response()
)
return handle_get_data_store_obj_w_code(
data_store_code = data_store_code,
for_type = for_type,
for_id = for_id,
commons = v3_commons,
)
# ### END ### API Data Store ### get_v3_data_store_obj_w_code() ###
# ### BEGIN ### API Data Store ### get_data_store_obj_w_code() ###
# NOTE: Adding some explanation because this is not quickly obvious how it fully works.
# The look up order starts with a required data_store_code. Then the first result that matches the most specific method. The for_type and for_id fields are not required. I think it makes the most sense to be a part of the URL path, not the GET params. Either should work with no problem though.

33
documentation/GUIDE__V3_FRONTEND_API.md
View File

@@ -136,4 +136,35 @@ V3 returns machine-readable error objects in `meta.details` for failures.
- `database_duplicate`: Non-unique value (Code 1062).
- `database_constraint`: Foreign key violation (Codes 1451, 1452).
- `database_schema`: Invalid column name (Codes 1054, 1146).
- `validation`: Pydantic validation failed (Check `details` for field-specific errors).
- `validation`: Pydantic validation failed (Check `details` for field-specific errors).
---
## 8. Data Store Cascading Lookup (V3)
V3 provides a specialized endpoint for retrieving configuration or content snippets by a human-friendly `code`. This uses a **hierarchical fallback** logic to find the best fit for the current context.
### A. The V3 Lookup Endpoint
**Path:** `GET /v3/data_store/code/{code}`
**Query Parameters:**
| Parameter | Type | Required | Description |
| :--- | :--- | :--- | :--- |
| `for_type` | String | No | Parent object type (e.g., `event`, `person`). |
| `for_id` | String | No | Parent object random ID. |
### B. Cascading Logic (Specificity)
The API automatically resolves the "best fit" record in the following order:
1. **Record Specific:** Matches `for_type` + `for_id` (and account).
2. **Account Specific:** Matches the `x-account-id` header.
3. **Global Default:** Matches `code` where `account_id` is `NULL`.
### C. Example Implementation
```ts
// GET /v3/data_store/code/event_launcher_main_info?for_type=event&for_id=nmBfuGFeR0k
export async function get_data_store_v3({ api_cfg, code, for_type, for_id }) {
const endpoint = `/v3/data_store/code/${code}`;
const params = { for_type, for_id };
return await get_object({ api_cfg, endpoint, params });
}
```

93
tests/e2e/test_e2e_v3_data_store_lookup.py Normal file
View File

@@ -0,0 +1,93 @@
import requests
import json
import argparse
# --- Configuration ---
BASE_URL = "https://dev-api.oneskyit.com"
AGENT_API_KEY = "IDF68Em5X4HTZlswRNgepQ"
# Default Contexts for Testing
CONTEXTS = {
"account_1": "_XY7DXtc9MY",
"account_5": "xFP7AhU8Zlc",
"account_22": "3Iid1aIRY5j",
"account_23": "nqOzejLCDXM",
"event_1358": "nmBfuGFeR0k"
}
def run_lookup(code, description, account_id=None, for_type=None, for_id=None, version="v3"):
"""
Performs a Data Store lookup and prints standardized results.
"""
print(f"[{version.upper()}] {description}")
headers = {
"X-Aether-API-Key": AGENT_API_KEY,
"Content-Type": "application/json"
}
if account_id:
headers["X-Account-ID"] = account_id
else:
headers["X-No-Account-ID"] = "bypass"
if version == "v3":
url = f"{BASE_URL}/v3/data_store/code/{code}"
params = {"for_type": for_type, "for_id": for_id}
response = requests.get(url, headers=headers, params=params)
else:
# Legacy Endpoint
if for_type and for_id:
url = f"{BASE_URL}/data_store/code/{code}/{for_type}/{for_id}"
else:
url = f"{BASE_URL}/data_store/code/{code}"
response = requests.get(url, headers=headers)
print(f" URL: {response.url}")
print(f" Status: {response.status_code}")
if response.status_code == 200:
data = response.json().get('data')
if data:
obj = data[0] if isinstance(data, list) else data
rec_id = obj.get('id') or obj.get('data_store_id') or obj.get('data_store_id_random')
print(f" Result: SUCCESS")
print(f" ID: {rec_id}")
print(f" Name: {obj.get('name')}")
else:
print(f" Result: NULL (No record found or validation failed)")
else:
print(f" Result: ERROR - {response.text[:200]}")
print("-" * 60)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Aether Data Store V3 Lookup Tester")
parser.add_argument("--code", default="event_launcher_main_info", help="The Data Store code to test")
args = parser.parse_args()
print(f"=== Aether Data Store Unified Tester ===")
print(f"Target: {BASE_URL}")
print(f"Code: {args.code}\n")
# 1. Global Context
run_lookup(args.code, "Scenario: Global Context (Bypass Account)")
# 2. Account 1 Context
run_lookup(args.code, "Scenario: Account 1 Context", account_id=CONTEXTS["account_1"])
# 3. Account 22 Context
run_lookup(args.code, "Scenario: Account 22 Context", account_id=CONTEXTS["account_22"])
# 4. Object Specific Context (Event 1358 - belongs to Account 1)
run_lookup(args.code, "Scenario: Event 1358 (under Account 1)",
account_id=CONTEXTS["account_1"],
for_type="event",
for_id=CONTEXTS["event_1358"])
# 5. Cross-Account Security Check (Event 1358 requested by Account 23)
run_lookup(args.code, "Scenario: Security Check (Event 1358 by Account 23 - SHOULD BE NULL)",
account_id=CONTEXTS["account_23"],
for_type="event",
for_id=CONTEXTS["event_1358"])
print("\nTests Complete.")