def fetch_tickers()

in src/backend/entrypoints/llm_backend/api/tickers.py [0:0]

• 37 lines of code
• 3 McCabe index (conditional complexity)

def fetch_tickers(request_model: TickersFetchRequestModel = Depends(), session: Session = Depends(get_session)):
    current_user = user.get()

    task = celery_app.signature(
        "fetch_tickers",
        kwargs={
            "data_provider": request_model.data_provider,
            "datasets": request_model.datasets,
            "periodicity": request_model.periodicity,
            "tradable_symbols_prompt": request_model.tradable_symbols_prompt,
            "supplementary_symbols_prompt": request_model.supplementary_symbols_prompt,
            "economic_indicators": request_model.economic_indicators,
            "dividend_fields": request_model.dividend_fields,
            "time_range": request_model.time_range,
        },
        queue="alerts_default",
        routing_key="alerts_default",
        headers=create_task_headers(session.session_id),
    )

    all_symbols, _, _ = parse_tickers(request_model.tradable_symbols_prompt, request_model.data_provider)

    num_symbols = sum(len(v) for v in all_symbols.values())

    work_units = num_symbols * sum(WORK_UNITS_PER_DATASET.get(ds, 0) for ds in request_model.datasets) + len(
        request_model.economic_indicators
    )

    task_id = str(uuid.uuid4())
    task_scheduled_timestamp = time.time()

    save_task_ws_meta_info(
        task_id,
        work_units=work_units,
        start_time=task_scheduled_timestamp,
    )

    session.add_action(task_id, task_scheduled_timestamp)
    session_manager_singleton.save(current_user.email, session)

    soft_time_limit = 10 * 60
    task.apply_async(soft_time_limit=soft_time_limit, time_limit=soft_time_limit + 20, task_id=task_id)

    return {
        "task_id": task_id,
    }