feat: add V3 action endpoint for event exhibit tracking export

- New router: app/routers/api_v3_actions_event_exhibit.py
  - GET /v3/action/event_exhibit/{exhibit_id}/tracking_export
  - Full V3 auth (x-aether-api-key + account context)
  - Multi-tenant ownership check via check_account_access
  - Permission gate: leads_api_access flag OR manager-level access
  - Returns CSV or XLSX file attachment (return_file=false for JSON)
  - Flattens responses_json custom Q&A columns; strips HTML from exhibitor_notes
  - Exports all records regardless of hidden/enabled state

- Registered in registry.py under prefix /v3/action/event_exhibit

- New E2E test: tests/e2e/test_e2e_v3_action_event_exhibit_tracking_export.py
  - 7/7 tests passing against dev-api.oneskyit.com

- Docs: GUIDE__AE_API_V3_for_Frontend.md — new Section 7 covering endpoint
  usage, columns, leads_api_access dual-purpose (3rd-party API + UI export gate)

- Docs: tests/README.md — added test to table and when-to-run matrix
This commit is contained in:
Scott Idem
2026-03-16 16:50:32 -04:00
parent 5f3ba1e03e
commit 29579fd9f1
5 changed files with 537 additions and 2 deletions

View File

@@ -0,0 +1,232 @@
import datetime
import logging
import os
import pathlib
import re
from typing import Optional
import pandas
from fastapi import APIRouter, Depends, HTTPException, Path, Query, status
from fastapi.responses import FileResponse
log = logging.getLogger(__name__)
from app.config import settings
from app.db_sql import redis_lookup_id_random, sql_select
from app.lib_api_crud_v3 import check_account_access
from app.lib_export import create_export_file, return_full_tmp_path
from app.lib_general_v3 import AccountContext, get_account_context
from app.methods.event_exhibit_tracking_methods import (
get_event_exhibit_tracking_rec_list,
load_event_exhibit_tracking_obj,
)
from app.models.response_models import mk_resp
"""
Aether API V3 - Event Exhibit Action Router
---------------------------------------------
Handles specialized actions for the Event Exhibit module, such as
exporting tracking (lead) data for exhibitors.
"""
router = APIRouter()
# --- Helpers ---
_HTML_TAG_RE = re.compile(r'<[^>]+>')
def _strip_html(text: Optional[str]) -> Optional[str]:
if not text:
return text
return _HTML_TAG_RE.sub('', text)
def _flatten_responses(responses_json: Optional[dict]) -> dict:
"""
Flatten responses_json into key→value pairs for CSV/Excel export.
New format: { question_code: { response: <value> } } → value = inner['response']
Legacy format: { label: <scalar> } → value = scalar
"""
if not responses_json:
return {}
flat = {}
for key, value in responses_json.items():
if isinstance(value, dict):
flat[key] = value.get('response')
else:
flat[key] = value
return flat
# --- Routes ---
@router.get('/{exhibit_id}/tracking_export')
async def export_exhibit_tracking(
exhibit_id: str = Path(..., min_length=11, max_length=22),
file_type: str = Query('CSV', regex=r'^(CSV|XLSX)$'),
return_file: bool = Query(True),
account: AccountContext = Depends(get_account_context),
):
"""
V3 Action: Export all tracking (lead capture) records for an exhibit.
Auth: Requires `leads_api_access == True` on the exhibit OR manager-level account access.
Returns a CSV or XLSX file attachment.
"""
# 1. Resolve random ID → internal integer
exhibit_int_id = redis_lookup_id_random(record_id_random=exhibit_id, table_name='event_exhibit')
if not exhibit_int_id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Exhibit not found.')
# 2. Load exhibit record for ownership + permission checks
exhibit_rec = sql_select(table_name='v_event_exhibit', record_id=exhibit_int_id)
if not exhibit_rec:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Exhibit not found.')
# 3. Multi-tenant ownership check
if not check_account_access(exhibit_rec, account, 'event_exhibit'):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail='Access denied: this exhibit belongs to a different account.',
)
# 4. Permission: leads_api_access flag OR manager-level access
if not exhibit_rec.get('leads_api_access') and not account.manager:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail='Access denied: leads API access is not enabled for this exhibit.',
)
# 5. Fetch all tracking records — no hidden/enabled filter, full export
tracking_rec_list = get_event_exhibit_tracking_rec_list(
event_exhibit_id=exhibit_int_id,
hidden='all',
enabled='all',
limit=1500,
)
if tracking_rec_list is False:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail='Failed to retrieve tracking records.',
)
# 6. Build export rows
data_rows = []
response_keys: list = [] # ordered unique custom-question column names
for rec in (tracking_rec_list or []):
tracking_obj = load_event_exhibit_tracking_obj(
event_exhibit_tracking_id=rec.get('event_exhibit_tracking_id'),
)
if not tracking_obj:
continue
# Apply exhibitor-side field overrides (override values replace badge defaults)
if tracking_obj.event_badge_full_name_override:
tracking_obj.event_badge_full_name = tracking_obj.event_badge_full_name_override
if tracking_obj.event_badge_professional_title_override:
tracking_obj.event_badge_professional_title = tracking_obj.event_badge_professional_title_override
if tracking_obj.event_badge_affiliations_override:
tracking_obj.event_badge_affiliations = tracking_obj.event_badge_affiliations_override
if tracking_obj.event_badge_email_override:
tracking_obj.event_badge_email = tracking_obj.event_badge_email_override
if tracking_obj.event_badge_location_override:
tracking_obj.event_badge_location = tracking_obj.event_badge_location_override
# Flatten custom Q&A responses and collect column keys (order-preserving dedup)
responses = _flatten_responses(tracking_obj.responses_json)
for key in responses:
if key not in response_keys:
response_keys.append(key)
row = {
'event_exhibit_tracking_id': tracking_obj.event_exhibit_tracking_id,
'created_on': tracking_obj.created_on,
'updated_on': tracking_obj.updated_on,
'event_exhibit_name': tracking_obj.event_exhibit_name,
'event_badge_full_name': tracking_obj.event_badge_full_name,
'event_badge_email': tracking_obj.event_badge_email,
'event_badge_professional_title': tracking_obj.event_badge_professional_title,
'event_badge_affiliations': tracking_obj.event_badge_affiliations,
'event_badge_location': tracking_obj.event_badge_location,
'event_badge_country': tracking_obj.event_badge_country,
'external_person_id': tracking_obj.external_person_id,
'exhibitor_notes': _strip_html(tracking_obj.exhibitor_notes),
'priority': tracking_obj.priority,
'enable': tracking_obj.enable,
'hide': tracking_obj.hide,
**responses,
}
data_rows.append(row)
# 7. Determine file format
export_type = 'Excel' if file_type == 'XLSX' else 'CSV'
content_type = (
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
if file_type == 'XLSX' else 'text/csv'
)
datetime_str = datetime.datetime.utcnow().strftime('%Y-%m-%d_%H%M')
filename = f'leads_export_{datetime_str}'
ext = '.xlsx' if export_type == 'Excel' else '.csv'
filename_w_ext = filename + ext
fixed_columns = [
'event_exhibit_tracking_id',
'created_on',
'updated_on',
'event_exhibit_name',
'event_badge_full_name',
'event_badge_email',
'event_badge_professional_title',
'event_badge_affiliations',
'event_badge_location',
'event_badge_country',
'external_person_id',
'exhibitor_notes',
'priority',
'enable',
'hide',
]
column_name_li = fixed_columns + response_keys
# 8. Handle empty result — write headers-only file
if not data_rows:
hosted_tmp_path = settings.FILES_PATH['hosted_tmp_root']
subdir = os.path.join(hosted_tmp_path, 'event_exhibit')
pathlib.Path(subdir).mkdir(parents=True, exist_ok=True)
full_path = os.path.join(subdir, filename_w_ext)
df = pandas.DataFrame(columns=fixed_columns)
if export_type == 'CSV':
df.to_csv(full_path, index=False)
else:
df.to_excel(full_path, index=False)
if return_file:
return FileResponse(path=full_path, filename=filename_w_ext, media_type=content_type)
return mk_resp(data=[], tmp_file_path=filename_w_ext)
# 9. Generate the export file
tmp_file_path = create_export_file(
data_dict_list=data_rows,
column_name_li=column_name_li,
subdir_path='event_exhibit',
filename=filename,
rm_id=False,
export_type=export_type,
)
if not tmp_file_path:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail='Failed to generate export file.',
)
if return_file:
full_path = return_full_tmp_path(full_tmp_path=tmp_file_path)
if not full_path:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail='Export file not found after creation.',
)
return FileResponse(path=full_path, filename=filename_w_ext, media_type=content_type)
return mk_resp(data=data_rows, tmp_file_path=tmp_file_path)

View File

@@ -7,7 +7,7 @@ from app.routers import (
event_device, event_exhibit, event_exhibit_tracking, event_file, event_importing,
event_location, event_person,
event_presentation, event_presenter, event_session,
flask_cfg, hosted_file, api_v3_actions_hosted_file, api_v3_actions_event_file, api_v3_actions_e_zoom, api_v3_actions_e_novi_mailman, lookup, lookup_v3,
flask_cfg, hosted_file, api_v3_actions_hosted_file, api_v3_actions_event_file, api_v3_actions_event_exhibit, api_v3_actions_e_zoom, api_v3_actions_e_novi_mailman, lookup, lookup_v3,
organization, page, person,
person_user, qr, site, site_domain, user,
util_email, websockets, websockets_redis, websockets_v3, e_confex, e_cvent, e_impexium, e_stripe
@@ -50,6 +50,7 @@ def setup_routers(app: FastAPI):
app.include_router(hosted_file.router, prefix='/hosted_file', tags=['Hosted File'])
app.include_router(api_v3_actions_hosted_file.router, prefix='/v3/action/hosted_file', tags=['Hosted File (V3 Actions)'])
app.include_router(api_v3_actions_event_file.router, prefix='/v3/action/event_file', tags=['Event File (V3 Actions)'])
app.include_router(api_v3_actions_event_exhibit.router, prefix='/v3/action/event_exhibit', tags=['Event Exhibit (V3 Actions)'])
app.include_router(api_v3_actions_e_zoom.router, prefix='/v3/action/e_zoom', tags=['Zoom Events (V3 Actions)'])
app.include_router(api_v3_actions_e_novi_mailman.router, prefix='/v3/action/e_novi_mailman', tags=['Novi-Mailman Bridge (V3 Actions)'])
app.include_router(lookup.router, prefix='/lu', tags=['Lookup'])

View File

@@ -154,8 +154,77 @@ Frontend guidance:
- These endpoints run synchronously and can take time for large inputs; for heavy or batch workloads use a queued job pattern instead.
- These endpoints may take time for large inputs. Prefer using `?background=true` to schedule work and receive a `202 Accepted` response for async processing. For heavy or batch workloads use a queued job pattern instead.
---
## 5. Troubleshooting 403 Forbidden
## 7. Event Exhibit Tracking Export (Leads Export)
Allows an exhibitor to download all lead-capture records for their exhibit as a CSV or XLSX file.
- **Method:** `GET`
- **Path:** `/v3/action/event_exhibit/{exhibit_id}/tracking_export`
- **Auth:** Standard V3 headers (`x-aether-api-key` + `x-account-id` or `?jwt=`)
### Query Parameters
| Parameter | Type | Default | Description |
| :--- | :--- | :--- | :--- |
| `file_type` | `CSV` \| `XLSX` | `CSV` | Output format. |
| `return_file` | bool | `true` | `true` → file download response. `false` → JSON body with row data. |
### Response
- `Content-Type: text/csv` (CSV) or `application/vnd.openxmlformats-officedocument.spreadsheetml.sheet` (XLSX)
- `Content-Disposition: attachment; filename="leads_export_<timestamp>.csv"`
- If there are no tracking records, a valid file with headers only is returned (not a 404).
### Columns Returned
Fixed columns (always present), followed by any custom question columns flattened from `responses_json`:
`event_exhibit_tracking_id`, `created_on`, `updated_on`, `event_exhibit_name`, `event_badge_full_name`, `event_badge_email`, `event_badge_professional_title`, `event_badge_affiliations`, `event_badge_location`, `event_badge_country`, `external_person_id`, `exhibitor_notes`, `priority`, `enable`, `hide`, `[custom question codes…]`
> **Note:** `exhibitor_notes` has HTML tags stripped automatically for clean CSV output.
### Permission Requirement — `leads_api_access`
> [!IMPORTANT]
> This endpoint enforces a **per-exhibit permission flag**. The `event_exhibit` record **must** have `leads_api_access = true` set in the database, OR the caller must have manager-level account access (JWT with `manager: true`).
>
> If `leads_api_access` is `false` or `null` on the exhibit, the API returns:
> ```json
> { "detail": "Access denied: leads API access is not enabled for this exhibit." }
> ```
> **Fix:** Enable the flag on the exhibit record via `PATCH /v3/crud/event_exhibit/{id}` with `{ "leads_api_access": true }`, or set it directly in the database/admin panel.
#### Dual purpose of `leads_api_access`
This flag serves two related but distinct roles:
1. **3rd-party API access (original intent):** Controls whether external systems (exhibitor apps, badge-scanning devices, etc.) are permitted to push or pull lead data for this exhibit via the API.
2. **UI export gate (new):** The frontend should read `leads_api_access` from the exhibit record and use it to show or hide the export/download button. Only render the button when the flag is `true` — this prevents users from triggering a request that will always 403.
The recommended pattern is to fetch the exhibit record first and gate the UI on this field before the user ever sees the export option. The API enforces the same check server-side as a safety net.
### Example Request
```ts
const resp = await fetch(
`https://dev-api.oneskyit.com/v3/action/event_exhibit/${exhibitId}/tracking_export?file_type=CSV&return_file=true`,
{
headers: {
'x-aether-api-key': API_KEY,
'x-account-id': accountId,
},
}
);
// resp is a file blob — use URL.createObjectURL() or trigger a download
const blob = await resp.blob();
const url = URL.createObjectURL(blob);
```
---
## 8. Troubleshooting 403 Forbidden
If you receive a 403 on a valid ID:
1. Verify `x-aether-api-key` is correct.

View File

@@ -28,6 +28,7 @@ These consolidated scripts are the primary verification tool for the V3 API.
| `test_e2e_v3_demo_parity.py` | **Demo Parity + Nested Create Regression**: Vision ID check for Badge, Exhibit, Tracking; nested create lifecycle (POST+DELETE) for `journal/journal_entry` and `event/event_session`; alias resolution. **Run after any model or nested-router change.** |
| `test_e2e_v3_action_event_file.py` | **Event Actions**: Specialized atomic upload and linking for event files. |
| `test_e2e_v3_action_zoom.py` | **Zoom Integration**: Verifies OAuth and ticket sync logic for Zoom Events. |
| `test_e2e_v3_action_event_exhibit_tracking_export.py` | **Exhibit Leads Export**: Auth/permission guards, CSV column structure, XLSX bytes, and `return_file` mode for the V3 tracking export action. |
| `test_e2e_v3_accounts.py` | CRUD verification for the core Account object. |
| `test_e2e_v3_schema.py` | Network verification of the V3 metadata discovery endpoint. |
| `test_e2e_agent_bridge.py` | Verifies container diagnostics and log streaming routes. |
@@ -52,6 +53,7 @@ Tests exist to be used — run the relevant suite whenever you touch backend cod
| Search / filter changes | `test_e2e_v3_search_engine.py` |
| Auth / account context changes | `test_e2e_v3_security_audit.py`, `test_e2e_v3_auth_security.py` |
| File upload / download changes | `test_e2e_v3_actions_file_lifecycle.py` |
| Event exhibit tracking export changes | `test_e2e_v3_action_event_exhibit_tracking_export.py` |
| Any backend change before frontend hand-off | All of the above |
---

View File

@@ -0,0 +1,231 @@
"""
E2E Test: V3 Action — Event Exhibit Tracking Export
Route: GET /v3/action/event_exhibit/{exhibit_id}/tracking_export
Tests:
1. Auth guard — rejected when API key is missing
2. Auth guard — rejected when account context is missing
3. Permission guard — rejected when leads_api_access is not enabled (without manager bypass)
4. Success (bypass) — CSV file returned with correct headers
5. Success (bypass) — XLSX file returned
6. Column structure — expected fixed columns present in CSV
7. 404 for a bogus exhibit ID
"""
import csv
import io
import sys
import time
import requests
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
BASE_URL = "https://dev-api.oneskyit.com/v3/action/event_exhibit"
API_KEY = "PMM4n50teUCaOMMTN8qOJA" # Agent API Key
# This exhibit is the stable demo record (from tests/README.md "event_exhibit")
# xK_9yEj1bQY is verified to exist in the demo environment (TARGETS list in demo_parity).
EXHIBIT_ID = "xK_9yEj1bQY"
BYPASS_HEADERS = {
"x-aether-api-key": API_KEY,
"x-no-account-id": "bypass",
}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def print_result(label: str, success: bool, message: str = ""):
status = "✅ PASS" if success else "❌ FAIL"
line = f" {status} | {label}"
if message:
line += f"{message}"
print(line)
return success
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_missing_api_key():
"""No auth at all → 403."""
resp = requests.get(f"{BASE_URL}/{EXHIBIT_ID}/tracking_export")
ok = resp.status_code == 403
print_result("Missing API key → 403", ok, f"got {resp.status_code}")
return ok
def test_missing_account_context():
"""API key present but no account context → 403."""
resp = requests.get(
f"{BASE_URL}/{EXHIBIT_ID}/tracking_export",
headers={"x-aether-api-key": API_KEY},
)
ok = resp.status_code == 403
print_result("Missing account context → 403", ok, f"got {resp.status_code}")
return ok
def test_leads_api_access_gate():
"""
A real (non-bypass) account that does NOT own this exhibit should be blocked.
Demo account '_XY7DXtc9MY' is used here; if it happens to own the exhibit and
have leads_api_access, this test will get a 200 — that's still acceptable data
(the endpoint is working). The main value is confirming no 500 error.
"""
resp = requests.get(
f"{BASE_URL}/{EXHIBIT_ID}/tracking_export",
headers={
"x-aether-api-key": API_KEY,
"x-account-id": "_XY7DXtc9MY", # Demo account
},
)
# Accept 403 (correct gate) or 200 (demo account owns exhibit with access enabled)
ok = resp.status_code in (200, 403)
note = "blocked (correct)" if resp.status_code == 403 else "allowed (demo account owns exhibit)"
print_result("leads_api_access gate", ok, f"got {resp.status_code}{note}")
return ok
def test_bogus_exhibit_id():
"""Non-existent exhibit ID → 404."""
resp = requests.get(
f"{BASE_URL}/AAAAAAAAAAA/tracking_export",
headers=BYPASS_HEADERS,
)
ok = resp.status_code == 404
print_result("Bogus exhibit ID → 404", ok, f"got {resp.status_code}")
return ok
def test_csv_export_bypass():
"""Bypass auth → CSV file returned with correct content-type and columns."""
resp = requests.get(
f"{BASE_URL}/{EXHIBIT_ID}/tracking_export",
params={"file_type": "CSV", "return_file": "true"},
headers=BYPASS_HEADERS,
)
if resp.status_code != 200:
print_result("CSV export (bypass)", False, f"got {resp.status_code}: {resp.text[:200]}")
return False
# Content-Type check
ct = resp.headers.get("content-type", "")
ct_ok = "text/csv" in ct
print_result("CSV content-type header", ct_ok, ct)
# Content-Disposition check
cd = resp.headers.get("content-disposition", "")
cd_ok = "attachment" in cd and ".csv" in cd
print_result("CSV content-disposition header", cd_ok, cd)
# Parse and check columns
try:
reader = csv.DictReader(io.StringIO(resp.text))
fieldnames = reader.fieldnames or []
expected_fixed = [
"event_exhibit_tracking_id",
"created_on",
"updated_on",
"event_exhibit_name",
"event_badge_full_name",
"event_badge_email",
"event_badge_professional_title",
"event_badge_affiliations",
"event_badge_location",
"event_badge_country",
"external_person_id",
"exhibitor_notes",
"priority",
"enable",
"hide",
]
missing = [c for c in expected_fixed if c not in fieldnames]
cols_ok = len(missing) == 0
print_result(
"CSV expected columns present",
cols_ok,
f"missing: {missing}" if missing else f"{len(fieldnames)} columns total",
)
rows = list(reader)
print_result("CSV parseable", True, f"{len(rows)} data rows")
except Exception as e:
print_result("CSV parse", False, str(e))
return False
return ct_ok and cd_ok and cols_ok
def test_xlsx_export_bypass():
"""Bypass auth → XLSX file returned with correct content-type."""
resp = requests.get(
f"{BASE_URL}/{EXHIBIT_ID}/tracking_export",
params={"file_type": "XLSX", "return_file": "true"},
headers=BYPASS_HEADERS,
)
if resp.status_code != 200:
print_result("XLSX export (bypass)", False, f"got {resp.status_code}: {resp.text[:200]}")
return False
ct = resp.headers.get("content-type", "")
ct_ok = "spreadsheetml" in ct or "openxmlformats" in ct or "octet-stream" in ct
print_result("XLSX content-type header", ct_ok, ct)
cd = resp.headers.get("content-disposition", "")
cd_ok = "attachment" in cd and ".xlsx" in cd
print_result("XLSX content-disposition header", cd_ok, cd)
# Basic magic-bytes check (XLSX starts with PK zip header)
magic_ok = resp.content[:2] == b"PK"
print_result("XLSX magic bytes (PK zip)", magic_ok, "")
return ct_ok and cd_ok and magic_ok
def test_return_file_false():
"""return_file=false → JSON response body instead of a file download."""
resp = requests.get(
f"{BASE_URL}/{EXHIBIT_ID}/tracking_export",
params={"file_type": "CSV", "return_file": "false"},
headers=BYPASS_HEADERS,
)
if resp.status_code != 200:
print_result("return_file=false", False, f"got {resp.status_code}: {resp.text[:200]}")
return False
ct = resp.headers.get("content-type", "")
json_ok = "json" in ct
print_result("return_file=false → JSON response", json_ok, ct)
return json_ok
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
if __name__ == "__main__":
print("=" * 60)
print("E2E: V3 Action — Event Exhibit Tracking Export")
print("=" * 60)
t_start = time.time()
results = [
test_missing_api_key(),
test_missing_account_context(),
test_leads_api_access_gate(),
test_bogus_exhibit_id(),
test_csv_export_bypass(),
test_xlsx_export_bypass(),
test_return_file_false(),
]
elapsed = time.time() - t_start
passed = sum(results)
total = len(results)
print()
print(f"Results: {passed}/{total} passed ({elapsed:.2f}s)")
sys.exit(0 if passed == total else 1)