Command: python -c "from pathlib import Path; import re; files=[Path(r'C:\Users\ash\Documents\ShopeeApi\app\main.py'),Path(r'C:\Users\ash\Documents\ShopeeApi\app\db.py'),Path(r'C:\Users\ash\Documents\ShopeeApi\app\shop_context.py')]; pats=re.compile(r'(api/shops|shop_registry|CREATE TABLE.*shop|marketplace|platform|def .*shop|active_shop)',re.I); [(print('FILE='+str(p)),[(print(str(i+1)+': '+line[:260])) for i,line in enumerate(p.read_text(encoding='utf-8',errors='replace').splitlines()) if pats.search(line)]) for p in files if p.exists()]"
Directory: projects
Status: SUCCESS
Exit code: 0
Cancel Job Rerun Command Refresh
FILE=C:\Users\ash\Documents\ShopeeApi\app\main.py
31: from app.shop_context import current_shop_id, reset_active_shop_id, set_active_shop_id
100: async def active_shop_context_middleware(request: Request, call_next):
115: token = set_active_shop_id(selected)
121: reset_active_shop_id(token)
297: platform: str = Field(default="SHOPEE", max_length=32)
748: def _selected_shop_id(value: int | str | None = None) -> int:
756: def _registry_scope_shop_id(value: int | str | None = None) -> int | None:
771: def _shop_records() -> list[dict[str, Any]]:
795: def _safe_token_status(shop_id: int) -> dict[str, Any]:
815: def _upsert_shop_record(shop_id: int, name: str) -> dict[str, Any]:
829: def _register_authorized_shop(client: ShopeeClient, saved: dict[str, Any]) -> tuple[dict[str, Any], str]:
1142: def shopee_auth_url():
1150: def legacy_shopee_auth():
1158: def legacy_shopee_auth_url():
1162: def shopee_callback(code: str | None = None, shop_id: int | None = None, main_account_id: int | None = None, merchant_id: int | None = None):
1184: async def api_shopee_push(
1213: def api_shopee_push_status():
1218: def api_webchat_messages(limit: int = 50, shop_id: int = 0):
1223: def api_shop_info():
1231: def legacy_shop_info():
1241: active_shop = int(current_shop_id() or settings.shop_id or 0)
1246: "active_shop_id": active_shop,
1250: "token_status": _safe_token_status(active_shop),
1255: active_shop = int(current_shop_id() or settings.shop_id or 0)
1257: "active_shop_id": active_shop,
1263: "token_status": _safe_token_status(active_shop),
1271: @app.get("/api/shops", dependencies=[Depends(require_admin)])
1272: def api_list_shops():
1273: return {"active_shop_id": current_shop_id(), "items": _shop_records()}
1276: @app.post("/api/shops", dependencies=[Depends(require_admin)])
1277: def api_upsert_shop(req: ShopUpsertRequest):
1283: @app.patch("/api/shops/{shop_id}", dependencies=[Depends(require_admin)])
1284: def api_patch_shop(shop_id: int, req: ShopUpsertRequest):
1292: @app.get("/api/shops/{shop_id}/auth-url", dependencies=[Depends(require_admin)])
1293: def api_shop_auth_url(shop_id: int):
1320: def _packing_sticker_profile(shop_id: int, profile_id: str = "") -> dict[str, Any]:
1339: def _packing_sticker_plan(req: PackingStickerRequest, shop_id: int) -> dict[str, Any]:
1581: platform=req.platform,
1612: platform=req.platform,
2017: "active_shop_id": current_shop_id(),
2158: def _start_arranged_awb_prepare(shop_id: int, documents: list[dict[str, Any]], source: str) -> dict[str, Any] | None:
2369: def _awb_branding_payload(shop_id: int) -> dict[str, Any]:
2384: def _awb_print_insert_payload(shop_id: int) -> dict[str, Any]:
2409: def _awb_remote_printer(shop_id: int, printer_name: str) -> tuple[dict[str, Any], str]:
2472: def _rebuild_cached_custom_awb(shop_id: int, order_sn: str, package_number: str) -> Path | None:
2950: def api_awb_branding_logo_from_shopee():
3079: def _label_profile_or_404(profile_id: str, shop_id: int) -> dict[str, Any]:
3269: def _scanner_catalog_revision(shop_id: int) -> str:
3325: def _scanner_forecast_revision(shop_id: int) -> str:
3336: def _scanner_snapshot_revision(shop_id: int) -> str:
3343: def _scanner_revisions(shop_id: int) -> dict[str, str]:
3712: def _scanner_alias_for_shop(shop_id: int, entry_id: int) -> dict[str, Any]:
4111: def api_scanner_batch_shopee_preview(batch_id: str):
4121: def api_scanner_batch_shopee_publish(batch_id: str, req: InventoryPublishRequest):
FILE=C:\Users\ash\Documents\ShopeeApi\app\db.py
74: def _shop_id(value: int | str | None = None) -> int:
135: def authenticate_scanner_device(token_sha256: str, shop_id: int | None = None) -> dict[str, Any] | None:
154: def list_scanner_devices(shop_id: int, limit: int = 100) -> list[dict[str, Any]]:
164: def revoke_scanner_device(shop_id: int, device_id: str) -> bool:
173: def enroll_operations_device(shop_id: int, device_id: str, token_sha256: str, device_name: str = "", app_version: str = "") -> dict[str, Any]:
188: def authenticate_operations_device(token_sha256: str, shop_id: int) -> dict[str, Any] | None:
204: def set_operations_primary_device(shop_id: int, device_id: str) -> dict[str, Any]:
222: def operations_alert_settings(shop_id: int) -> dict[str, Any]:
228: def acknowledge_instant_delivery(shop_id: int, order_sn: str, device_id: str) -> dict[str, Any]:
246: def instant_delivery_acknowledgements(shop_id: int, order_sns: list[str]) -> dict[str, dict[str, Any]]:
399: "platform='SHOPEE'",
634: def _awb_document_relpath(shop_id: int, kind: str, order_sn: Any, package_number: Any, source_hash: str | None = None, logo_hash: str | None = None) -> str:
916: def _load_perfume_registry_maps(conn: sqlite3.Connection, shop_id: int | None = None) -> dict[str, dict[str, str]]:
921: SELECT id, shop_id, platform, source_mode, listing_name, variant_name, match_key, canonical_name, source, created_at, updated_at
923: WHERE platform='SHOPEE' AND shop_id IN (0, ?)
1180: INSERT INTO perfume_aliases(platform, listing_name, listing_norm, canonical_name, image_1, image_2, image_3, source, updated_at)
1182: ON CONFLICT(platform, listing_norm) DO UPDATE SET
1189: (row.get("platform") or "SHOPEE", listing, norm, canonical, row.get("image_1") or "", row.get("image_2") or "", row.get("image_3") or "", "Diamond Perfume Data.xlsx", now),
1263: ("SELECT DISTINCT canonical_name FROM perfume_aliases WHERE platform='SHOPEE' AND TRIM(COALESCE(canonical_name,''))<>''", ()),
1286: platform="SHOPEE",
1308: platform: str = "SHOPEE",
1323: INSERT INTO perfume_name_registry(shop_id, platform, source_mode, listing_name, variant_name, match_key, canonical_name, source, created_at, updated_at)
1325: ON CONFLICT(shop_id, platform, source_mode, match_key) DO UPDATE SET
1332: (int(shop_id), platform or "SHOPEE", listing, variant, match_key, canonical, source or "catalog_auto", now, now),
1337: INSERT INTO perfume_name_registry(shop_id, platform, source_mode, listing_name, variant_name, match_key, canonical_name, source, created_at, updated_at)
1339: ON CONFLICT(shop_id, platform, source_mode, match_key) DO NOTHING
1341: (int(shop_id), platform or "SHOPEE", listing, variant, match_key, canonical, source or "catalog_auto", now, now),
1345: SELECT id, shop_id, platform, source_mode, listing_name, variant_name, match_key, canonical_name, source, created_at, updated_at
1347: WHERE shop_id=? AND platform=? AND source_mode='canonical' AND match_key=?
1351: (int(shop_id), platform or "SHOPEE", match_key),
1525: def _required_positive_shop_id(value: int | str) -> int:
1542: def _ensure_shop_sync_policy_conn(
1565: def _seed_shop_onboarding(conn: sqlite3.Connection, now: str) -> None:
1609: def start_shop_onboarding(
1655: def get_shop_sync_policy(shop_id: int | str) -> dict[str, Any] | None:
1662: def get_shop_onboarding(shop_id: int | str) -> dict[str, Any] | None:
1669: def update_shop_onboarding(
1707: def shop_onboarding_ready(shop_id: int | str) -> bool:
1752: def get_scanner_forecast_revision(shop_id: int | str) -> dict[str, Any]:
1773: CREATE TABLE IF NOT EXISTS shops (
1945: CREATE TABLE IF NOT EXISTS shopee_ads_shop_status (
1954: CREATE TABLE IF NOT EXISTS shopee_ads_daily_performance (
1974: CREATE TABLE IF NOT EXISTS shopee_ads_campaign_daily_performance (
2028: CREATE TABLE IF NOT EXISTS shopee_returns (
2048: CREATE TABLE IF NOT EXISTS shopee_return_items (
2135: platform TEXT NOT NULL DEFAULT 'SHOPEE',
2144: PRIMARY KEY (platform, listing_norm)
2159: platform TEXT NOT NULL DEFAULT 'SHOPEE',
2171: CREATE UNIQUE INDEX IF NOT EXISTS idx_perfume_name_registry_scope_key ON perfume_name_registry(shop_id, platform, source_mode, match_key);
2420: CREATE TABLE IF NOT EXISTS shop_sync_policies (
2431: CREATE TABLE IF NOT EXISTS shop_onboarding (
2447: CREATE TABLE IF NOT EXISTS shop_scanner_revisions (
2629: CREATE TABLE IF NOT EXISTS awb_shop_branding (
2659: CREATE TABLE IF NOT EXISTS awb_shop_print_insert (
2671: CREATE TABLE IF NOT EXISTS shopee_auth_states (
2745: CREATE TABLE IF NOT EXISTS shop_product_mappings (
2904: CREATE TABLE IF NOT EXISTS shopee_push_events (
2925: CREATE TABLE IF NOT EXISTS shopee_webchat_messages (
3190: where = ["pnr.platform='SHOPEE'"]
3214: SELECT pnr.id, pnr.shop_id, pnr.platform, pnr.source_mode, pnr.listing_name, pnr.variant_name, pnr.match_key, pnr.canonical_name, pnr.source, pnr.created_at, pnr.updated_at,
3250: "pnr.platform='SHOPEE'",
3265: SELECT pnr.id, pnr.shop_id, pnr.platform, pnr.source_mode, pnr.listing_name, pnr.variant_name, pnr.match_key, pnr.canonical_name, pnr.source, pnr.created_at, pnr.updated_at,
3328: "platform": str(row["platform"] or "SHOPEE"),
3358: "platform": "SHOPEE",
3402: platform: str = "SHOPEE",
3441: SET shop_id=?, platform=?, source_mode=?, listing_name=?, variant_name=?, match_key=?, canonical_name=?, sku=?, associate_code=?, capacity_ml=?, source=?, updated_at=?
3444: (sid, platform or "SHOPEE", mode, listing, variant, match_key, canonical, sku_value, associate_value, capacity_ml, source or "manual", now, int(entry_id)),
3447: "SELECT id, shop_id, platform, source_mode, listing_name, variant_name, match_key, canonical_name, sku, associate_code, capacity_ml, source, created_at, updated_at FROM perfume_name_registry WHERE id=?",
3454: INSERT INTO perfume_name_registry(shop_id, platform, source_mode, listing_name, variant_name, match_key, canonical_name, sku, associate_code, capacity_ml, source, created_at, updated_at)
3456: ON CONFLICT(shop_id, platform, source_mode, match_key) DO UPDATE SET
3466: (sid, platform or "SHOPEE", mode, listing, variant, match_key, canonical, sku_value, associate_value, capacity_ml, source or "manual", now, now),
3470: SELECT id, shop_id, platform, source_mode, listing_name, variant_name, match_key, canonical_name, sku, associate_code, capacity_ml, source, created_at, updated_at
3472: WHERE shop_id=? AND platform=? AND source_mode=? AND match_key=?
3476: (sid, platform or "SHOPEE", mode, match_key),
3485: platform=platform or "SHOPEE",
3498: "platform": platform or "SHOPEE",
3511: def delete_perfume_registry_entry(entry_id: int, shop_id: int | None = None) -> dict[str, Any]:
3514: row = conn.execute("SELECT id, shop_id, platform, source_mode, listing_name, variant_name, match_key, canonical_name, source, created_at, updated_at FROM perfume_name_registry WHERE id=?", (int(entry_id),)).fetchone()
3701: def save_shop_product_mapping(
3792: def deactivate_shop_product_mapping(shop_id: int, mapping_id: str, reason: str = "") -> dict[str, Any]:
3816: def bulk_confirm_shop_product_mappings(shop_id: int, mapping_ids: list[str]) -> dict[str, Any]:
3999: def list_shop_product_mapping_queue(
4158: def list_scanner_name_products(shop_id: int, search: str = "") -> list[dict[str, Any]]:
4163: where = ["pnr.platform='SHOPEE'", "pnr.shop_id IN (0, ?)"]
4221: WHERE platform='SHOPEE'
4341: def replace_canonical_stickers(shop_id: int, canonical_name: str, sticker_codes: list[str], actor: str = "admin") -> dict[str, Any]:
4407: WHERE shop_id=? AND platform='SHOPEE' AND source_mode='canonical'
4422: shop_id, platform, source_mode, listing_name, variant_name,
4439: SELECT id, shop_id, platform, source_mode, listing_name, variant_name,
4443: WHERE shop_id=? AND platform='SHOPEE' AND source_mode='canonical'
4465: def rename_scanner_canonical(shop_id: int, old_name: str, new_name: str) -> dict[str, Any]:
4506: def create_sync_job(shop_id: int, module: str, params: dict[str, Any], progress_total: int = 0, actor: str = "admin") -> str:
4524: def enqueue_shopee_push_event(
4557: def recover_shopee_push_events() -> int:
4567: def claim_shopee_push_event(max_attempts: int) -> dict[str, Any] | None:
4624: def list_webchat_messages(shop_id: int, limit: int = 50) -> list[dict[str, Any]]:
4640: def complete_shopee_push_event(
4665: def retry_shopee_push_event(event_id: str, error: str, *, attempts: int, max_attempts: int) -> None:
4686: def shopee_push_status() -> dict[str, Any]:
4718: def get_active_job(module: str, shop_id: int | str) -> dict[str, Any] | None:
4732: def list_jobs(limit: int = 30, shop_id: int | None = None) -> list[dict[str, Any]]:
4741: def has_active_job(module: str | None = None, shop_id: int | None = None) -> bool:
4764: def get_order_sync_states(order_sns: list[str], shop_id: int | None = None) -> dict[str, dict[str, Any]]:
4789: def get_order_package_numbers(shop_id: int, order_sns: list[str]) -> dict[str, set[str]]:
4923: def cleanup_stale_sync_jobs(shop_id: int | None = None, module: str | None = None) -> int:
4950: def get_last_completed_sync_end_ts(module: str, shop_id: int | None = None) -> int | None:
4990: def get_awb_shop_branding(shop_id: int | None = None) -> dict[str, Any] | None:
4997: def touch_awb_shop_branding(shop_id: int | None = None) -> dict[str, Any] | None:
5007: def list_registered_shop_ids() -> list[int]:
5013: def is_registered_shop_id(shop_id: int | str) -> bool:
5024: def upsert_shop_identity(shop_id: int, shop_name: str) -> None:
5041: def get_shop_name(shop_id: int) -> str:
5048: def upsert_awb_shop_branding(
5077: def delete_awb_shop_branding(shop_id: int | None = None) -> None:
5083: def invalidate_branded_awb_for_shop(shop_id: int | None = None) -> int:
5101: def get_awb_document(shop_id: int, order_sn: str, package_number: str | None = "") -> dict[str, Any] | None:
5112: def get_awb_custom_document(shop_id: int, order_sn: str, package_number: str | None = "") -> dict[str, Any] | None:
5191: def invalidate_custom_awb_for_shop(shop_id: int) -> int:
5205: def get_awb_shop_print_insert(shop_id: int) -> dict[str, Any] | None:
5215: def upsert_awb_shop_print_insert(
5264: def delete_awb_shop_print_insert(shop_id: int) -> None:
5271: def upsert_awb_document(shop_id: int, order_sn: str, package_number: str | None = "", **fields: Any) -> dict[str, Any]:
5364: def mark_awb_documents_printed(shop_id: int, document_keys: list[dict[str, Any]]) -> int:
5396: def list_printable_awb_orders(shop_id: int | None = None, printed_filter: str = "all", limit: int = 300) -> list[dict[str, Any]]:
5503: def _upsert_order_items(conn: sqlite3.Connection, shop_id: int, order_sn: str, items: list[Any], now: str) -> None:
5527: def _upsert_order_packages(conn: sqlite3.Connection, shop_id: int, order_sn: str, packages: list[Any], now: str) -> None:
5562: def upsert_orders(shop_id: int, orders: list[dict[str, Any]]) -> int:
5644: def save_order_raw(job_id: str | None, shop_id: int, order_sn: str | None, endpoint: str, payload: dict[str, Any]) -> None:
5652: def save_order_error(job_id: str | None, shop_id: int, order_sn: str | None, stage: str, error: str, payload: Any | None = None) -> None:
5759: def upsert_order_escrow(shop_id: int, order_sn: str, payload: dict[str, Any]) -> None:
5805: def upsert_escrow_releases(shop_id: int, rows: list[dict[str, Any]]) -> int:
5842: def rebuild_order_financials(shop_id: int) -> int:
5857: def upsert_order_logistics(shop_id: int, order_sn: str, package_number: str | None, endpoint: str, payload: dict[str, Any]) -> None:
6012: def _upsert_product_variations(conn: sqlite3.Connection, shop_id: int, item_id: int, product: dict[str, Any], now: str, product_currency: str | None = None) -> int | None:
6017: def upsert_products(shop_id: int, products: list[dict[str, Any]]) -> int:
6136: def get_order_status(order_sn: str, shop_id: int | None = None) -> str:
6177: def get_cancelled_order(order_sn: str, shop_id: int | None = None) -> dict[str, Any] | None:
6187: def get_order(order_sn: str, shop_id: int | None = None) -> dict[str, Any] | None:
6526: alias_rows = list(conn.execute("SELECT listing_name, listing_norm, canonical_name FROM perfume_aliases WHERE platform='SHOPEE'").fetchall())
6533: if str(seed.get("platform") or "SHOPEE").upper() != "SHOPEE":
6706: def _inventory_runtime_storage(conn: sqlite3.Connection, shop_id: int) -> dict[str, Any]:
6761: alias_rows = list(conn.execute("SELECT listing_name, listing_norm, canonical_name FROM perfume_aliases WHERE platform='SHOPEE'").fetchall())
6763: if str(seed.get("platform") or "SHOPEE").upper() != "SHOPEE":
7273: alias_rows = list(conn.execute("SELECT listing_name, listing_norm, canonical_name FROM perfume_aliases WHERE platform='SHOPEE'").fetchall())
7825: def canonical_name_is_registered(name: str, shop_id: int | None = None) -> bool:
7861: def _ready_stock_row(conn: Any, shop_id: int, canonical_name: str, size_ml: int) -> dict[str, Any]:
7988: def list_ready_stock(shop_id: int, search: str = "", size_ml: int = 0) -> list[dict[str, Any]]:
8010: def list_ready_stock_movements(shop_id: int, limit: int = 50, order_sn: str = "") -> list[dict[str, Any]]:
8232: def list_label_print_agents(shop_id: int | None = None) -> list[dict[str, Any]]:
8344: def get_label_printer_profile(profile_id: str, shop_id: int | None = None) -> dict[str, Any] | None:
8361: def list_label_printer_profiles(shop_id: int | None = None) -> list[dict[str, Any]]:
8379: def set_label_profile_calibration(profile_id: str, shop_id: int, status: str) -> dict[str, Any] | None:
8463: def get_label_print_job(job_id: str, shop_id: int | None = None) -> dict[str, Any] | None:
8481: def get_label_job_by_idempotency(shop_id: int, idempotency_key: str) -> dict[str, Any] | None:
8490: def list_label_print_jobs(shop_id: int | None = None, limit: int = 100) -> list[dict[str, Any]]:
8581: def cancel_label_print_job(job_id: str, shop_id: int) -> bool:
8665: def get_remote_print_job(job_id: str, shop_id: int | None = None) -> dict[str, Any] | None:
8675: def list_remote_print_jobs(shop_id: int | None = None, limit: int = 100) -> list[dict[str, Any]]:
8690: def cancel_remote_print_job(job_id: str, shop_id: int) -> bool:
8793: def get_awb_direct_print_settings(shop_id: int | None = None) -> dict[str, Any] | None:
8861: def list_inventory_products(shop_id: int | None = None, search: str = "") -> list[dict[str, Any]]:
8877: def get_inventory_product_by_code(shop_id: int, product_code: str) -> dict[str, Any] | None:
8891: def get_inventory_product_by_legacy_family(shop_id: int, product_code: str) -> dict[str, Any] | None:
8966: def get_inventory_container(shop_id: int, container_code: str) -> dict[str, Any] | None:
9027: def _next_inventory_product_code(conn: sqlite3.Connection, shop_id: int) -> str:
9093: WHERE shop_id=? AND platform='SHOPEE' AND source_mode='canonical'
9143: def create_inventory_batch(shop_id: int, batch_type: str, note: str = "", client_batch_id: str = "") -> dict[str, Any]:
9166: def get_inventory_batch(shop_id: int, batch_id: str) -> dict[str, Any] | None:
9175: def list_inventory_batches(shop_id: int, limit: int = 100) -> list[dict[str, Any]]:
9194: def acquire_inventory_batch_lock(shop_id: int, batch_id: str, device_id: str, lease_seconds: int = 120) -> dict[str, Any]:
9227: def heartbeat_inventory_batch_lock(shop_id: int, batch_id: str, device_id: str, lease_token: str, lease_seconds: int = 120) -> dict[str, Any]:
9251: def release_inventory_batch_lock(shop_id: int, batch_id: str, device_id: str, lease_token: str) -> dict[str, Any]:
9375: def rename_inventory_batch(shop_id: int, batch_id: str, note: str) -> dict[str, Any]:
9389: def delete_inventory_batch(shop_id: int, batch_id: str) -> None:
9510: def list_inventory_batch_items(shop_id: int, batch_id: str) -> list[dict[str, Any]]:
9600: def update_inventory_batch_item(shop_id: int, item_id: str, **changes: Any) -> dict[str, Any]:
9647: def delete_inventory_batch_item(shop_id: int, item_id: str) -> None:
9664: def close_inventory_batch(shop_id: int, batch_id: str, status: str) -> dict[str, Any]:
9677: def set_inventory_batch_commit(shop_id: int, batch_id: str, action: str, request_id: str) -> dict[str, Any]:
9691: def update_inventory_batch_shopee(
9724: def clear_inventory_batch_items(shop_id: int, batch_id: str) -> None:
9729: def create_inventory_replace_session(shop_id: int, note: str = "") -> dict[str, Any]:
9741: def get_inventory_replace_session(shop_id: int, session_id: str) -> dict[str, Any] | None:
9750: def list_inventory_replace_sessions(shop_id: int, limit: int = 50) -> list[dict[str, Any]]:
9759: def list_inventory_replace_items(shop_id: int, session_id: str) -> list[dict[str, Any]]:
9775: def add_inventory_replace_item(shop_id: int, session_id: str, item: dict[str, Any]) -> dict[str, Any]:
9801: def update_inventory_replace_item(shop_id: int, session_id: str, item_id: str, item: dict[str, Any]) -> dict[str, Any]:
9831: def delete_inventory_replace_item(shop_id: int, session_id: str, item_id: str) -> bool:
9843: def set_inventory_replace_session_status(shop_id: int, session_id: str, status: str) -> dict[str, Any]:
9857: def list_inventory_label_queue(shop_id: int, status: str = "queued", limit: int = 500) -> list[dict[str, Any]]:
9869: def mark_inventory_label_queue_job(shop_id: int, queue_ids: list[str], job_id: str) -> int:
9882: def commit_inventory_replace_session(shop_id: int, session_id: str, *, request_id: str, device_id: str = "", note: str = "") -> dict[str, Any]:
9934: def _next_container_code(conn: sqlite3.Connection, shop_id: int, product_code: str) -> tuple[str, int]:
10124: def list_inventory_movements(shop_id: int, limit: int = 200) -> list[dict[str, Any]]:
FILE=C:\Users\ash\Documents\ShopeeApi\app\shop_context.py
9: _ACTIVE_SHOP_ID: ContextVar[int] = ContextVar("active_shopee_shop_id", default=0)
12: def current_shop_id(default: int | None = None) -> int:
13: value = int(_ACTIVE_SHOP_ID.get() or 0)
26: def set_active_shop_id(shop_id: int | str | None) -> Token:
31: return _ACTIVE_SHOP_ID.set(value if value > 0 else 0)
34: def reset_active_shop_id(token: Token) -> None:
35: _ACTIVE_SHOP_ID.reset(token)
39: def use_shop(shop_id: int | str | None) -> Iterator[int]:
40: token = set_active_shop_id(shop_id)
44: reset_active_shop_id(token)