def query_news_articles()

in app/services/bigquery.py [0:0]

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

def query_news_articles() -> bf.DataFrame:
    """
    Query BigQuery to retrieve news articles from the last 6 months.

    Returns:
        bf.DataFrame: A BigFrames DataFrame containing the news articles.
    """
    logger.info("🔍 Starting BigQuery news articles query process")
    client: bigquery.Client = connect_to_bigquery()
    bf.options.bigquery.client = client
    bf.options.bigquery.location = GBQ_LOCATION
    bf.options.bigquery.project = GBQ_PROJECT_ID

    columns: List[str] = [
        "sys_id",
        "sys_updated_on",
        "rich_content_html",
        "sys_created_on",
        "content_template_value",
        "content_template_link",
        "subheadline",
        "news_end_date",
        "news_start_date",
        "headline",
        "_fivetran_synced",
        "_fivetran_deleted",
        "thumbnail",
        "rich_content_components",
        "rich_content_css"
    ]

    # Calculate date 6 months ago
    six_months_ago = (datetime.now() - timedelta(days=180)).strftime('%Y-%m-%d')

    # Check if required values exist before constructing the query
    if not all([GBQ_PROJECT_ID, GBQ_DATASET, GBQ_NEWS_TABLE]):
        raise ValueError("Missing required BigQuery configuration values")

    query: str = f"""
        SELECT      {', '.join(columns)}
        FROM        `{GBQ_PROJECT_ID}.{GBQ_DATASET}.{GBQ_NEWS_TABLE}`
        WHERE       news_start_date >= '{six_months_ago}'
                    AND (_fivetran_deleted IS NULL OR _fivetran_deleted = FALSE)
        LIMIT       {GBQ_MAX_RESULTS}
    """

    logger.info(f"\n📝 News Query: {query}\n")

    df: bf.DataFrame = bf.read_gbq(query)
    logger.info(f"✨ News query executed, retrieved {len(df)} rows")

    return df