relay-knowledge 1.1.17

Graph-database-based knowledge graph project.
Documentation
use std::collections::{BTreeMap, BTreeSet};

use super::arguments::{
    extract_python_router_argument, python_prefix_argument, trim_one_trailing_paren,
};

#[cfg(test)]
#[path = "routers_tests.rs"]
mod tests;

#[derive(Clone)]
pub(super) struct PythonRouterInfo {
    pub(super) local_prefix: String,
    pub(super) mount_prefixes: BTreeSet<String>,
    pub(super) framework: String,
    pub(super) mount_required: bool,
    pub(super) cross_file_mount_candidate: bool,
}

pub(super) fn parse_python_router_prefix(line: &str) -> Option<(String, PythonRouterInfo)> {
    let (left, right) = line.split_once('=')?;
    let router_name = python_assignment_name(left)?;
    if router_name.is_empty()
        || !router_name
            .chars()
            .all(|character| character.is_ascii_alphanumeric() || character == '_')
    {
        return None;
    }
    let router_info = if let Some(args) = python_call_arguments(right, "APIRouter(") {
        let local_prefix = python_prefix_argument(args, "prefix");
        PythonRouterInfo {
            cross_file_mount_candidate: python_router_name_is_cross_file_candidate(
                &router_name,
                &local_prefix,
            ),
            local_prefix,
            mount_prefixes: BTreeSet::new(),
            framework: "fastapi".to_owned(),
            mount_required: true,
        }
    } else if python_call_arguments(right, "FastAPI(").is_some() {
        PythonRouterInfo {
            local_prefix: String::new(),
            mount_prefixes: BTreeSet::new(),
            framework: "fastapi".to_owned(),
            mount_required: false,
            cross_file_mount_candidate: false,
        }
    } else {
        let args = python_call_arguments(right, "Blueprint(")?;
        let local_prefix = python_prefix_argument(args, "url_prefix");
        PythonRouterInfo {
            cross_file_mount_candidate: python_router_name_is_cross_file_candidate(
                &router_name,
                &local_prefix,
            ),
            local_prefix,
            mount_prefixes: BTreeSet::new(),
            framework: "flask".to_owned(),
            mount_required: true,
        }
    };

    Some((router_name, router_info))
}

fn python_call_arguments<'a>(source: &'a str, marker: &str) -> Option<&'a str> {
    let start = source.find(marker)? + marker.len();
    Some(trim_one_trailing_paren(&source[start..]))
}

pub(super) fn merge_python_router_declaration(
    routers: &mut BTreeMap<String, PythonRouterInfo>,
    router_name: String,
    mut router_info: PythonRouterInfo,
) {
    if let Some(existing) = routers.remove(&router_name) {
        router_info.mount_prefixes = existing.mount_prefixes;
    }
    routers.insert(router_name, router_info);
}

fn python_router_name_is_cross_file_candidate(router_name: &str, local_prefix: &str) -> bool {
    (router_name == "router" && !local_prefix.is_empty())
        || (!matches!(router_name, "bp" | "blueprint")
            && (router_name.ends_with("_router")
                || router_name.ends_with("_blueprint")
                || router_name.ends_with("Router")
                || router_name.ends_with("Blueprint")))
}

pub(super) fn apply_python_include_router_prefix(
    statement: &str,
    routers: &mut BTreeMap<String, PythonRouterInfo>,
) -> bool {
    let Some(paren_pos) = statement.find(".include_router(") else {
        return false;
    };
    let args = trim_one_trailing_paren(&statement[paren_pos + ".include_router(".len()..]);
    let Some(router_name) = extract_python_router_argument(args, "router") else {
        return false;
    };
    let prefix = python_prefix_argument(args, "prefix");
    let router_info = routers
        .entry(router_name)
        .or_insert_with(|| PythonRouterInfo {
            local_prefix: String::new(),
            mount_prefixes: BTreeSet::new(),
            framework: "fastapi".to_owned(),
            mount_required: true,
            cross_file_mount_candidate: false,
        });
    router_info.mount_prefixes.insert(prefix);
    router_info.framework = "fastapi".to_owned();
    true
}

pub(super) fn apply_python_register_blueprint_prefix(
    statement: &str,
    routers: &mut BTreeMap<String, PythonRouterInfo>,
) -> bool {
    let Some(paren_pos) = statement.find(".register_blueprint(") else {
        return false;
    };
    let args = trim_one_trailing_paren(&statement[paren_pos + ".register_blueprint(".len()..]);
    let Some(blueprint_name) = extract_python_router_argument(args, "blueprint") else {
        return false;
    };
    let prefix = python_prefix_argument(args, "url_prefix");
    let router_info = routers
        .entry(blueprint_name)
        .or_insert_with(|| PythonRouterInfo {
            local_prefix: String::new(),
            mount_prefixes: BTreeSet::new(),
            framework: "flask".to_owned(),
            mount_required: true,
            cross_file_mount_candidate: false,
        });
    router_info.mount_prefixes.insert(prefix);
    router_info.framework = "flask".to_owned();
    true
}

pub(super) fn route_framework(
    func_part: &str,
    receiver_name: Option<&str>,
    routers: &BTreeMap<String, PythonRouterInfo>,
) -> String {
    if let Some(receiver_name) = receiver_name {
        if let Some(router_info) = routers.get(receiver_name) {
            return router_info.framework.clone();
        }
    }
    if func_part.ends_with(".api_route") {
        return "fastapi".to_owned();
    }
    "flask".to_owned()
}

fn python_assignment_name(left: &str) -> Option<String> {
    let name = left
        .trim()
        .split_once(':')
        .map_or(left.trim(), |(name, _)| name.trim());
    if name.is_empty() {
        return None;
    }
    Some(name.to_owned())
}