main() — fastapi Function Reference

Architecture documentation for the main() function in notify_translations.py from the fastapi codebase.

Function python FastAPI Applications calls 4

Entity Profile

FastAPI→ Applications→ main() — fastapi Function Reference

Dependency Diagram

graph TD
  aff23c90_a22b_d9fe_46af_57f96d115b4e["main()"]
  97ea90ba_5175_e861_5c41_e18772cfce41["notify_translations.py"]
  aff23c90_a22b_d9fe_46af_57f96d115b4e -->|defined in| 97ea90ba_5175_e861_5c41_e18772cfce41
  db3757db_3aa7_c0de_3404_156cfb8b4378["get_graphql_translation_discussions()"]
  aff23c90_a22b_d9fe_46af_57f96d115b4e -->|calls| db3757db_3aa7_c0de_3404_156cfb8b4378
  d02f1c49_91bd_4d52_5264_714ea3daf1e4["get_graphql_translation_discussion_comments()"]
  aff23c90_a22b_d9fe_46af_57f96d115b4e -->|calls| d02f1c49_91bd_4d52_5264_714ea3daf1e4
  7b21b141_8eaf_f1b4_7d31_def0d718bfe6["create_comment()"]
  aff23c90_a22b_d9fe_46af_57f96d115b4e -->|calls| 7b21b141_8eaf_f1b4_7d31_def0d718bfe6
  ab7596ec_b849_0364_85c9_f9c3182b25bd["update_comment()"]
  aff23c90_a22b_d9fe_46af_57f96d115b4e -->|calls| ab7596ec_b849_0364_85c9_f9c3182b25bd
  style aff23c90_a22b_d9fe_46af_57f96d115b4e fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

scripts/notify_translations.py lines 306–428

def main() -> None:
    settings = Settings()
    if settings.debug:
        logging.basicConfig(level=logging.DEBUG)
    else:
        logging.basicConfig(level=logging.INFO)
    logging.debug(f"Using config: {settings.model_dump_json()}")
    g = Github(settings.github_token.get_secret_value())
    repo = g.get_repo(settings.github_repository)
    if not settings.github_event_path.is_file():
        raise RuntimeError(
            f"No github event file available at: {settings.github_event_path}"
        )
    contents = settings.github_event_path.read_text("utf-8")
    github_event = PartialGitHubEvent.model_validate_json(contents)
    logging.info(f"Using GitHub event: {github_event}")
    number = (
        github_event.pull_request and github_event.pull_request.number
    ) or settings.number
    if number is None:
        raise RuntimeError("No PR number available")

    # Avoid race conditions with multiple labels
    sleep_time = random.random() * 10  # random number between 0 and 10 seconds
    logging.info(
        f"Sleeping for {sleep_time} seconds to avoid "
        "race conditions and multiple comments"
    )
    time.sleep(sleep_time)

    # Get PR
    logging.debug(f"Processing PR: #{number}")
    pr = repo.get_pull(number)
    label_strs = {label.name for label in pr.get_labels()}
    langs = []
    for label in label_strs:
        if label.startswith("lang-") and not label == lang_all_label:
            langs.append(label[5:])
    logging.info(f"PR #{pr.number} has labels: {label_strs}")
    if not langs or lang_all_label not in label_strs:
        logging.info(f"PR #{pr.number} doesn't seem to be a translation PR, skipping")
        sys.exit(0)

    # Generate translation map, lang ID to discussion
    discussions = get_graphql_translation_discussions(settings=settings)
    lang_to_discussion_map: dict[str, AllDiscussionsDiscussionNode] = {}
    for discussion in discussions:
        for edge in discussion.labels.edges:
            label = edge.node.name
            if label.startswith("lang-") and not label == lang_all_label:
                lang = label[5:]
                lang_to_discussion_map[lang] = discussion
    logging.debug(f"Using translations map: {lang_to_discussion_map}")

    # Messages to create or check
    new_translation_message = f"Good news everyone! 😉 There's a new translation PR to be reviewed: #{pr.number} by @{pr.user.login}. 🎉 This requires 2 approvals from native speakers to be merged. 🤓"
    done_translation_message = f"~There's a new translation PR to be reviewed: #{pr.number} by @{pr.user.login}~ Good job! This is done. 🍰☕"

    # Normally only one language, but still
    for lang in langs:
        if lang not in lang_to_discussion_map:
            log_message = f"Could not find discussion for language: {lang}"
            logging.error(log_message)
            raise RuntimeError(log_message)
        discussion = lang_to_discussion_map[lang]
        logging.info(
            f"Found a translation discussion for language: {lang} in discussion: #{discussion.number}"
        )

        already_notified_comment: Union[Comment, None] = None
        already_done_comment: Union[Comment, None] = None

        logging.info(
            f"Checking current comments in discussion: #{discussion.number} to see if already notified about this PR: #{pr.number}"
        )
        comments = get_graphql_translation_discussion_comments(
            settings=settings, discussion_number=discussion.number
        )
        for comment in comments:
            if new_translation_message in comment.body:
                already_notified_comment = comment