use std::collections::{HashMap, HashSet};
use libcst_native::{
AssignTargetExpression, CompoundStatement, Element, Expression, ImportNames, Module,
NameOrAttribute, OrElse, SmallStatement, Statement, Suite,
};
pub struct Imports {
table: HashMap<String, String>,
relative_levels: HashMap<String, usize>,
dynamic: bool,
}
impl Imports {
pub fn build(module: &Module) -> Self {
let mut table: HashMap<String, String> = HashMap::new();
let mut relative_levels: HashMap<String, usize> = HashMap::new();
let mut dynamic = false;
for stmt in &module.body {
collect_stmt(stmt, &mut table, &mut relative_levels, &mut dynamic);
}
Self {
table,
relative_levels,
dynamic,
}
}
pub fn resolve(&self, local: &str) -> Option<&str> {
self.table.get(local).map(|s| s.as_str())
}
pub fn is_relative(&self, local: &str) -> bool {
self.relative_levels.contains_key(local)
}
pub fn relative_level(&self, local: &str) -> Option<usize> {
self.relative_levels.get(local).copied()
}
pub fn has_dynamic(&self) -> bool {
self.dynamic
}
}
fn collect_stmt(
stmt: &Statement,
table: &mut HashMap<String, String>,
relative_levels: &mut HashMap<String, usize>,
dynamic: &mut bool,
) {
match stmt {
Statement::Simple(line) => {
for small in &line.body {
collect_small(small, table, relative_levels, dynamic);
}
}
Statement::Compound(c) => collect_compound(c, table, relative_levels, dynamic),
}
}
fn collect_suite(
suite: &Suite,
table: &mut HashMap<String, String>,
relative_levels: &mut HashMap<String, usize>,
dynamic: &mut bool,
) {
match suite {
Suite::IndentedBlock(b) => {
for stmt in &b.body {
collect_stmt(stmt, table, relative_levels, dynamic);
}
}
Suite::SimpleStatementSuite(s) => {
for small in &s.body {
collect_small(small, table, relative_levels, dynamic);
}
}
}
}
fn collect_compound(
c: &CompoundStatement,
table: &mut HashMap<String, String>,
relative_levels: &mut HashMap<String, usize>,
dynamic: &mut bool,
) {
match c {
CompoundStatement::FunctionDef(d) => {
collect_suite(&d.body, table, relative_levels, dynamic)
}
CompoundStatement::ClassDef(d) => collect_suite(&d.body, table, relative_levels, dynamic),
CompoundStatement::If(i) => {
collect_suite(&i.body, table, relative_levels, dynamic);
if let Some(orelse) = &i.orelse {
collect_orelse(orelse, table, relative_levels, dynamic);
}
}
CompoundStatement::For(f) => {
collect_suite(&f.body, table, relative_levels, dynamic);
if let Some(e) = &f.orelse {
collect_suite(&e.body, table, relative_levels, dynamic);
}
}
CompoundStatement::While(w) => {
collect_suite(&w.body, table, relative_levels, dynamic);
if let Some(e) = &w.orelse {
collect_suite(&e.body, table, relative_levels, dynamic);
}
}
CompoundStatement::Try(t) => {
collect_suite(&t.body, table, relative_levels, dynamic);
for h in &t.handlers {
collect_suite(&h.body, table, relative_levels, dynamic);
}
if let Some(e) = &t.orelse {
collect_suite(&e.body, table, relative_levels, dynamic);
}
if let Some(e) = &t.finalbody {
collect_suite(&e.body, table, relative_levels, dynamic);
}
}
CompoundStatement::TryStar(t) => {
collect_suite(&t.body, table, relative_levels, dynamic);
for h in &t.handlers {
collect_suite(&h.body, table, relative_levels, dynamic);
}
if let Some(e) = &t.orelse {
collect_suite(&e.body, table, relative_levels, dynamic);
}
if let Some(e) = &t.finalbody {
collect_suite(&e.body, table, relative_levels, dynamic);
}
}
CompoundStatement::With(w) => collect_suite(&w.body, table, relative_levels, dynamic),
CompoundStatement::Match(m) => {
for case in &m.cases {
collect_suite(&case.body, table, relative_levels, dynamic);
}
}
}
}
fn collect_orelse(
orelse: &OrElse,
table: &mut HashMap<String, String>,
relative_levels: &mut HashMap<String, usize>,
dynamic: &mut bool,
) {
match orelse {
OrElse::Elif(elif) => {
collect_suite(&elif.body, table, relative_levels, dynamic);
if let Some(inner) = &elif.orelse {
collect_orelse(inner, table, relative_levels, dynamic);
}
}
OrElse::Else(e) => collect_suite(&e.body, table, relative_levels, dynamic),
}
}
fn collect_small(
small: &SmallStatement,
table: &mut HashMap<String, String>,
relative_levels: &mut HashMap<String, usize>,
dynamic: &mut bool,
) {
match small {
SmallStatement::Import(imp) => {
for alias in &imp.names {
let path = noa_to_string(&alias.name);
if path == "importlib" || path.starts_with("importlib.") {
*dynamic = true;
}
let local = if let Some(asname) = &alias.asname {
ate_to_string(&asname.name)
} else {
root_component(&path)
};
table.insert(local, path);
}
}
SmallStatement::ImportFrom(from) => {
let level = from.relative.len();
let module_path = from
.module
.as_ref()
.map(|m| noa_to_string(m))
.unwrap_or_default();
if module_path == "importlib" || module_path.starts_with("importlib.") {
*dynamic = true;
}
let ImportNames::Aliases(aliases) = &from.names else {
return;
};
for alias in aliases {
let name = noa_to_string(&alias.name);
if name == "__import__" {
*dynamic = true;
}
let full = if module_path.is_empty() {
name.clone()
} else {
format!("{module_path}.{name}")
};
let local = if let Some(asname) = &alias.asname {
ate_to_string(&asname.name)
} else {
name
};
if level > 0 {
relative_levels.insert(local.clone(), level);
}
table.insert(local, full);
}
}
_ => {}
}
}
fn noa_to_string(noa: &NameOrAttribute) -> String {
match noa {
NameOrAttribute::N(n) => n.value.to_owned(),
NameOrAttribute::A(a) => {
let mut parts: Vec<String> = Vec::new();
collect_attr_parts_expr(&a.value, &mut parts);
parts.push(a.attr.value.to_owned());
parts.join(".")
}
}
}
fn collect_attr_parts_expr(expr: &Expression, out: &mut Vec<String>) {
match expr {
Expression::Name(n) => out.push(n.value.to_owned()),
Expression::Attribute(a) => {
collect_attr_parts_expr(&a.value, out);
out.push(a.attr.value.to_owned());
}
_ => {}
}
}
fn ate_to_string(ate: &AssignTargetExpression) -> String {
match ate {
AssignTargetExpression::Name(n) => n.value.to_owned(),
_ => String::new(),
}
}
fn root_component(path: &str) -> String {
path.split('.').next().unwrap_or(path).to_owned()
}
pub fn module_bindings(module: &Module) -> HashSet<String> {
let mut out = HashSet::new();
for stmt in &module.body {
match stmt {
Statement::Simple(line) => {
for small in &line.body {
match small {
SmallStatement::Assign(a) => {
for target in &a.targets {
collect_target_names(&target.target, &mut out);
}
}
SmallStatement::AnnAssign(a) => {
collect_target_names(&a.target, &mut out);
}
_ => {}
}
}
}
Statement::Compound(c) => match c {
CompoundStatement::FunctionDef(f) => {
out.insert(f.name.value.to_owned());
}
CompoundStatement::ClassDef(c) => {
out.insert(c.name.value.to_owned());
}
_ => {}
},
}
}
out
}
pub(crate) fn collect_target_names(target: &AssignTargetExpression, out: &mut HashSet<String>) {
match target {
AssignTargetExpression::Name(n) => {
out.insert(n.value.to_owned());
}
AssignTargetExpression::Tuple(t) => {
for el in &t.elements {
collect_element_names(el, out);
}
}
AssignTargetExpression::List(l) => {
for el in &l.elements {
collect_element_names(el, out);
}
}
AssignTargetExpression::StarredElement(s) => collect_expr_target_names(&s.value, out),
AssignTargetExpression::Attribute(_) | AssignTargetExpression::Subscript(_) => {}
}
}
pub(crate) fn collect_element_names(el: &Element, out: &mut HashSet<String>) {
match el {
Element::Simple { value, .. } => collect_expr_target_names(value, out),
Element::Starred(s) => collect_expr_target_names(&s.value, out),
}
}
pub(crate) fn collect_expr_target_names(expr: &Expression, out: &mut HashSet<String>) {
match expr {
Expression::Name(n) => {
out.insert(n.value.to_owned());
}
Expression::Tuple(t) => {
for el in &t.elements {
collect_element_names(el, out);
}
}
Expression::List(l) => {
for el in &l.elements {
collect_element_names(el, out);
}
}
Expression::StarredElement(s) => collect_expr_target_names(&s.value, out),
_ => {}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn build_str(src: &str) -> Imports {
Imports::build(&libcst_native::parse_module(src, None).unwrap())
}
#[test]
fn resolves_import_forms() {
let i = build_str("import os\nimport numpy as np\nfrom subprocess import run\n");
assert_eq!(i.resolve("os"), Some("os"));
assert_eq!(i.resolve("np"), Some("numpy"));
assert_eq!(i.resolve("run"), Some("subprocess.run"));
}
#[test]
fn resolves_dotted_import_without_alias() {
let i = build_str("import a.b.c\n");
assert_eq!(i.resolve("a"), Some("a.b.c"));
assert_eq!(i.resolve("a.b.c"), None);
}
#[test]
fn resolves_from_import_with_alias() {
let i = build_str("from m import n as p\n");
assert_eq!(i.resolve("p"), Some("m.n"));
assert_eq!(i.resolve("n"), None);
}
#[test]
fn from_import_star_does_not_crash() {
let i = build_str("from os.path import *\n");
assert_eq!(i.resolve("join"), None);
assert!(!i.has_dynamic());
}
#[test]
fn detects_importlib_dynamic() {
let i = build_str("import importlib\n");
assert!(i.has_dynamic());
}
#[test]
fn detects_from_importlib_dynamic() {
let i = build_str("from importlib import import_module\n");
assert!(i.has_dynamic());
}
#[test]
fn detects_importlib_submodule_dynamic() {
assert!(build_str("import importlib.util\n").has_dynamic());
assert!(build_str("from importlib.util import find_spec\n").has_dynamic());
}
#[test]
fn no_dynamic_for_normal_imports() {
let i = build_str("import os\nfrom sys import path\n");
assert!(!i.has_dynamic());
}
#[test]
fn resolves_function_local_imports() {
let i = build_str("def f():\n import subprocess\n subprocess.run(c, shell=True)\n");
assert_eq!(i.resolve("subprocess"), Some("subprocess"));
}
#[test]
fn resolves_nested_class_method_imports_and_dynamic() {
let i = build_str(
"class C:\n def m(self):\n from subprocess import run\n import importlib\n",
);
assert_eq!(i.resolve("run"), Some("subprocess.run"));
assert!(i.has_dynamic());
}
#[test]
fn is_relative_detects_leading_dot_imports() {
let i = build_str("from .utils import helper\nfrom . import sibling\nimport os\n");
assert!(i.is_relative("helper"), "helper must be relative");
assert!(i.is_relative("sibling"), "sibling must be relative");
assert!(!i.is_relative("os"), "os must not be relative");
assert!(!i.is_relative("unknown"), "unknown must not be relative");
}
#[test]
fn module_bindings_collects_top_level_only() {
let src = "\
import config\n\
_counter = 0\n\
shared_map = {}\n\
A, B = 1, 2\n\
[x, y] = [3, 4]\n\
def helper():\n inner_local = 1\n return inner_local\n\
class Box:\n pass\n";
let module = libcst_native::parse_module(src, None).unwrap();
let mb = module_bindings(&module);
for name in [
"_counter",
"shared_map",
"A",
"B",
"x",
"y",
"helper",
"Box",
] {
assert!(
mb.contains(name),
"expected module binding `{name}`, got {mb:?}"
);
}
assert!(
!mb.contains("inner_local"),
"function-body local leaked into module_bindings"
);
assert!(
!mb.contains("config"),
"imported name leaked into module_bindings"
);
}
}