File: C:/Users/fred/anaconda3/Lib/site-packages/rope/contrib/autoimport/sqlite.py
"""AutoImport module for rope."""
import contextlib
import json
from hashlib import sha256
import secrets
import re
import sqlite3
import sys
import warnings
from collections import OrderedDict
from concurrent.futures import Future, ProcessPoolExecutor, as_completed
from datetime import datetime
from itertools import chain
from pathlib import Path
from threading import local
from typing import Generator, Iterable, Iterator, List, Optional, Set, Tuple
from rope.base import exceptions, libutils, resourceobserver, taskhandle, versioning
from rope.base.project import Project
from rope.base.resources import Resource
from rope.contrib.autoimport import models
from rope.contrib.autoimport.defs import (
ModuleFile,
Name,
NameType,
Package,
PackageType,
SearchResult,
Source,
)
from rope.contrib.autoimport.parse import get_names
from rope.contrib.autoimport.utils import (
get_files,
get_modname_from_path,
get_package_tuple,
sort_and_deduplicate,
sort_and_deduplicate_tuple,
)
from rope.refactor import importutils
def get_future_names(
packages: List[Package], underlined: bool, job_set: taskhandle.BaseJobSet
) -> Generator[Future, None, None]:
"""Get all names as futures."""
with ProcessPoolExecutor() as executor:
for package in packages:
for module in get_files(package, underlined):
job_set.started_job(module.modname)
job_set.increment()
yield executor.submit(get_names, module, package)
def filter_packages(
packages: Iterable[Package], underlined: bool, existing: List[str]
) -> Iterable[Package]:
"""Filter list of packages to parse."""
if underlined:
def filter_package(package: Package) -> bool:
return package.name not in existing
else:
def filter_package(package: Package) -> bool:
return package.name not in existing and not package.name.startswith("_")
return filter(filter_package, packages)
_deprecated_default: bool = object() # type: ignore
class AutoImport:
"""A class for finding the module that provides a name.
This class maintains a cache of global names in python modules.
Note that this cache is not accurate and might be out of date.
"""
connection: sqlite3.Connection
memory: bool
project: Project
project_package: Package
underlined: bool
def __init__(
self,
project: Project,
observe: bool = True,
underlined: bool = False,
memory: bool = _deprecated_default,
):
"""Construct an AutoImport object.
Parameters
___________
project : rope.base.project.Project
the project to use for project imports
observe : bool
if true, listen for project changes and update the cache.
underlined : bool
If `underlined` is `True`, underlined names are cached, too.
memory:
If true, don't persist to disk
DEPRECATION NOTICE: The default value will change to use an on-disk
database by default in the future. If you want to use an in-memory
database, you need to pass `memory=True` explicitly:
autoimport = AutoImport(..., memory=True)
"""
self.project = project
project_package = get_package_tuple(project.root.pathlib, project)
assert project_package is not None
assert project_package.path is not None
self.project_package = project_package
self.underlined = underlined
self.memory = memory
if memory is _deprecated_default:
self.memory = True
warnings.warn(
"The default value for `AutoImport(memory)` argument will "
"change to use an on-disk database by default in the future. "
"If you want to use an in-memory database, you need to pass "
"`AutoImport(memory=True)` explicitly.",
DeprecationWarning,
)
self.thread_local = local()
self.connection = self.create_database_connection(
project=project,
memory=memory,
)
self._setup_db()
if observe:
observer = resourceobserver.ResourceObserver(
changed=self._changed, moved=self._moved, removed=self._removed
)
project.add_observer(observer)
@classmethod
def create_database_connection(
cls,
*,
project: Optional[Project] = None,
memory: bool = False,
) -> sqlite3.Connection:
"""
Create an sqlite3 connection
project : rope.base.project.Project
the project to use for project imports
memory : bool
if true, don't persist to disk
"""
def calculate_project_hash(data: str) -> str:
return sha256(data.encode()).hexdigest()
if not memory and project is None:
raise Exception("if memory=False, project must be provided")
if memory or project is None or project.ropefolder is None:
# Allows the in-memory db to be shared across threads
# See https://www.sqlite.org/inmemorydb.html
project_hash: str
if project is None:
project_hash = secrets.token_hex()
elif project.ropefolder is None:
project_hash = calculate_project_hash(project.address)
else:
project_hash = calculate_project_hash(project.ropefolder.real_path)
return sqlite3.connect(
f"file:rope-{project_hash}:?mode=memory&cache=shared", uri=True
)
else:
return sqlite3.connect(project.ropefolder.pathlib / "autoimport.db")
@property
def connection(self):
"""
Creates a new connection if called from a new thread.
This makes sure AutoImport can be shared across threads.
"""
if not hasattr(self.thread_local, "connection"):
self.thread_local.connection = self.create_database_connection(
project=self.project,
memory=self.memory,
)
return self.thread_local.connection
@connection.setter
def connection(self, value: sqlite3.Connection):
self.thread_local.connection = value
def _setup_db(self):
models.Metadata.create_table(self.connection)
version_hash = list(
self._execute(models.Metadata.objects.select("version_hash"))
)
current_version_hash = versioning.calculate_version_hash(self.project)
if not version_hash or version_hash[0][0] != current_version_hash:
self.clear_cache()
def import_assist(self, starting: str):
"""
Find modules that have a global name that starts with `starting`.
For a more complete list, use the search or search_full methods.
Parameters
__________
starting : str
what all the names should start with
Return
__________
Return a list of ``(name, module)`` tuples
"""
results = self._execute(
models.Name.import_assist.select("name", "module", "source"), (starting,)
).fetchall()
return sort_and_deduplicate_tuple(
results
) # Remove duplicates from multiple occurrences of the same item
def search(self, name: str, exact_match: bool = False) -> List[Tuple[str, str]]:
"""
Search both modules and names for an import string.
This is a simple wrapper around search_full with basic sorting based on Source.
Returns a sorted list of import statement, modname pairs
"""
results: List[Tuple[str, str, int]] = [
(statement, import_name, source)
for statement, import_name, source, type in self.search_full(
name, exact_match
)
]
return sort_and_deduplicate_tuple(results)
def search_full(
self,
name: str,
exact_match: bool = False,
ignored_names: Optional[Set[str]] = None,
) -> Generator[SearchResult, None, None]:
"""
Search both modules and names for an import string.
Parameters
__________
name: str
Name to search for
exact_match: bool
If using exact_match, only search for that name.
Otherwise, search for any name starting with that name.
ignored_names : Set[str]
Will ignore any names in this set
Return
__________
Unsorted Generator of SearchResults. Each is guaranteed to be unique.
"""
if ignored_names is None:
ignored_names = set()
results = set(self._search_name(name, exact_match))
results = results.union(self._search_module(name, exact_match))
for result in results:
if result.name not in ignored_names:
yield result
def _search_name(
self, name: str, exact_match: bool = False
) -> Generator[SearchResult, None, None]:
"""
Search both names for available imports.
Returns the import statement, import name, source, and type.
"""
if not exact_match:
name = name + "%" # Makes the query a starts_with query
for import_name, module, source, name_type in self._execute(
models.Name.search_by_name_like.select("name", "module", "source", "type"),
(name,),
):
yield (
SearchResult(
f"from {module} import {import_name}",
import_name,
source,
name_type,
)
)
def _search_module(
self, name: str, exact_match: bool = False
) -> Generator[SearchResult, None, None]:
"""
Search both modules for available imports.
Returns the import statement, import name, source, and type.
"""
if not exact_match:
name = name + "%" # Makes the query a starts_with query
for module, source in self._execute(
models.Name.search_submodule_like.select("module", "source"), (name,)
):
parts = module.split(".")
import_name = parts[-1]
remaining = parts[0]
for part in parts[1:-1]:
remaining += "."
remaining += part
yield (
SearchResult(
f"from {remaining} import {import_name}",
import_name,
source,
NameType.Module.value,
)
)
for module, source in self._execute(
models.Name.search_module_like.select("module", "source"), (name,)
):
if "." in module:
continue
yield SearchResult(
f"import {module}", module, source, NameType.Module.value
)
def get_modules(self, name) -> List[str]:
"""Get the list of modules that have global `name`."""
results = self._execute(
models.Name.search_by_name.select("module", "source"), (name,)
).fetchall()
return sort_and_deduplicate(results)
def get_all_names(self) -> List[str]:
"""Get the list of all cached global names."""
return self._execute(models.Name.objects.select("name")).fetchall()
def _dump_all(self) -> Tuple[List[Name], List[Package]]:
"""Dump the entire database."""
name_results = self._execute(models.Name.objects.select_star()).fetchall()
package_results = self._execute(models.Package.objects.select_star()).fetchall()
return name_results, package_results
def generate_cache(
self,
resources: Optional[List[Resource]] = None,
underlined: bool = False,
task_handle: taskhandle.BaseTaskHandle = taskhandle.DEFAULT_TASK_HANDLE,
):
"""Generate global name cache for project files.
If `resources` is a list of `rope.base.resource.File`, only
those files are searched; otherwise all python modules in the
project are cached.
"""
if resources is None:
resources = self.project.get_python_files()
job_set = task_handle.create_jobset(
"Generating autoimport cache", len(resources)
)
self._execute(
models.Package.delete_by_package_name, (self.project_package.name,)
)
futures = []
with ProcessPoolExecutor() as executor:
for file in resources:
job_set.started_job(f"Working on {file.path}")
module = self._resource_to_module(file, underlined)
futures.append(executor.submit(get_names, module, self.project_package))
for future in as_completed(futures):
self._add_names(future.result())
job_set.finished_job()
self.connection.commit()
def generate_modules_cache(
self,
modules: Optional[List[str]] = None,
task_handle: taskhandle.BaseTaskHandle = taskhandle.DEFAULT_TASK_HANDLE,
single_thread: bool = False,
underlined: Optional[bool] = None,
):
"""
Generate global name cache for external modules listed in `modules`.
If no modules are provided, it will generate a cache for every module available.
This method searches in your sys.path and configured python folders.
Do not use this for generating your own project's internal names,
use generate_resource_cache for that instead.
"""
underlined = self.underlined if underlined is None else underlined
packages: List[Package] = (
self._get_available_packages()
if modules is None
else list(self._get_packages_from_modules(modules))
)
existing = self._get_packages_from_cache()
packages = list(filter_packages(packages, underlined, existing))
if not packages:
return
self._add_packages(packages)
job_set = task_handle.create_jobset("Generating autoimport cache", 0)
if single_thread:
for package in packages:
for module in get_files(package, underlined):
job_set.started_job(module.modname)
for name in get_names(module, package):
self._add_name(name)
job_set.finished_job()
else:
for future_name in as_completed(
get_future_names(packages, underlined, job_set)
):
self._add_names(future_name.result())
job_set.finished_job()
self.connection.commit()
def _get_packages_from_modules(self, modules: List[str]) -> Iterator[Package]:
for modname in modules:
package = self._find_package_path(modname)
if package is None:
continue
yield package
def update_module(self, module: str):
"""Update a module in the cache, or add it if it doesn't exist."""
self._del_if_exist(module)
self.generate_modules_cache([module])
def close(self):
"""Close the autoimport database."""
self.connection.commit()
self.connection.close()
def get_name_locations(self, name):
"""Return a list of ``(resource, lineno)`` tuples."""
result = []
modules = self._execute(
models.Name.search_by_name_like.select("module"), (name,)
).fetchall()
for module in modules:
with contextlib.suppress(exceptions.ModuleNotFoundError):
module_name = module[0]
if module_name.startswith(f"{self.project_package.name}."):
module_name = ".".join(module_name.split("."))
pymodule = self.project.get_module(module_name)
if name in pymodule:
pyname = pymodule[name]
module, lineno = pyname.get_definition_location()
if module is not None:
resource = module.get_module().get_resource()
if resource is not None and lineno is not None:
result.append((resource, lineno))
return result
def clear_cache(self):
"""Clear all entries in global-name cache.
It might be a good idea to use this function before
regenerating global names.
"""
with self.connection:
self._execute(models.Name.objects.drop_table())
self._execute(models.Package.objects.drop_table())
self._execute(models.Metadata.objects.drop_table())
models.Name.create_table(self.connection)
models.Package.create_table(self.connection)
models.Metadata.create_table(self.connection)
data = (
versioning.calculate_version_hash(self.project),
json.dumps(versioning.get_version_hash_data(self.project)),
datetime.utcnow().isoformat(),
)
assert models.Metadata.columns == [
"version_hash",
"hash_data",
"created_at",
]
self._execute(models.Metadata.objects.insert_into(), data)
self.connection.commit()
def find_insertion_line(self, code):
"""Guess at what line the new import should be inserted."""
match = re.search(r"^(def|class)\s+", code)
if match is not None:
code = code[: match.start()]
try:
pymodule = libutils.get_string_module(self.project, code)
except exceptions.ModuleSyntaxError:
return 1
testmodname = "__rope_testmodule_rope"
importinfo = importutils.NormalImport(((testmodname, None),))
module_imports = importutils.get_module_imports(self.project, pymodule)
module_imports.add_import(importinfo)
code = module_imports.get_changed_source()
offset = code.index(testmodname)
lineno = code.count("\n", 0, offset) + 1
return lineno
def update_resource(
self, resource: Resource, underlined: bool = False, commit: bool = True
):
"""Update the cache for global names in `resource`."""
underlined = underlined if underlined else self.underlined
module = self._resource_to_module(resource, underlined)
self._del_if_exist(module_name=module.modname, commit=False)
for name in get_names(module, self.project_package):
self._add_name(name)
if commit:
self.connection.commit()
def _changed(self, resource):
if not resource.is_folder():
self.update_resource(resource)
def _moved(self, resource: Resource, newresource: Resource):
if not resource.is_folder():
modname = self._resource_to_module(resource).modname
self._del_if_exist(modname)
self.update_resource(newresource)
def _del_if_exist(self, module_name, commit: bool = True):
self._execute(models.Name.delete_by_module_name, (module_name,))
if commit:
self.connection.commit()
def _get_python_folders(self) -> List[Path]:
def filter_folders(folder: Path) -> bool:
return folder.is_dir() and folder.as_posix() != "/usr/bin"
folders = self.project.get_python_path_folders()
folder_paths = filter(filter_folders, map(Path, folders))
return list(OrderedDict.fromkeys(folder_paths))
def _safe_iterdir(self, folder: Path):
dirs = folder.iterdir()
while True:
try:
yield next(dirs)
except PermissionError:
pass
except StopIteration:
break
def _get_available_packages(self) -> List[Package]:
packages: List[Package] = [
Package(module, Source.BUILTIN, None, PackageType.BUILTIN)
for module in sys.builtin_module_names
]
for folder in self._get_python_folders():
for package in self._safe_iterdir(folder):
package_tuple = get_package_tuple(package, self.project)
if package_tuple is None:
continue
packages.append(package_tuple)
return packages
def _add_packages(self, packages: List[Package]):
data = [(p.name, str(p.path)) for p in packages]
self._executemany(models.Package.objects.insert_into(), data)
def _get_packages_from_cache(self) -> List[str]:
existing: List[str] = list(
chain(*self._execute(models.Package.objects.select_star()).fetchall())
)
existing.append(self.project_package.name)
return existing
def _removed(self, resource):
if not resource.is_folder():
modname = self._resource_to_module(resource).modname
self._del_if_exist(modname)
def _add_future_names(self, names: Future):
self._add_names(names.result())
@staticmethod
def _convert_name(name: Name) -> tuple:
return (
name.name,
name.modname,
name.package,
name.source.value,
name.name_type.value,
)
def _add_names(self, names: Iterable[Name]):
if names is not None:
self._executemany(
models.Name.objects.insert_into(),
[self._convert_name(name) for name in names],
)
def _add_name(self, name: Name):
self._execute(models.Name.objects.insert_into(), self._convert_name(name))
def _find_package_path(self, target_name: str) -> Optional[Package]:
if target_name in sys.builtin_module_names:
return Package(target_name, Source.BUILTIN, None, PackageType.BUILTIN)
for folder in self._get_python_folders():
for package in self._safe_iterdir(folder):
package_tuple = get_package_tuple(package, self.project)
if package_tuple is None:
continue
name, source, package_path, package_type = package_tuple
if name == target_name:
return package_tuple
return None
def _resource_to_module(
self, resource: Resource, underlined: bool = False
) -> ModuleFile:
assert self.project_package.path
underlined = underlined if underlined else self.underlined
resource_path: Path = resource.pathlib
# The project doesn't need its name added to the path,
# since the standard python file layout accounts for that
# so we set add_package_name to False
resource_modname: str = get_modname_from_path(
resource_path, self.project_package.path, add_package_name=False
)
return ModuleFile(
resource_path,
resource_modname,
underlined,
resource_path.name == "__init__.py",
)
def _execute(self, query: models.FinalQuery, *args, **kwargs):
assert isinstance(query, models.FinalQuery)
return self.connection.execute(query._query, *args, **kwargs)
def _executemany(self, query: models.FinalQuery, *args, **kwargs):
assert isinstance(query, models.FinalQuery)
return self.connection.executemany(query._query, *args, **kwargs)