mirror of
https://github.com/qdrant/landing_page.git
synced 2026-09-26 22:48:30 +02:00
103 lines
2.9 KiB
Python
103 lines
2.9 KiB
Python
import ast
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from .base import (
|
|
CompileResult,
|
|
Language,
|
|
copy_template,
|
|
generic_shorten,
|
|
trim_commonpath,
|
|
)
|
|
|
|
RE_IMPORTS = re.compile(
|
|
r"^from qdrant_client import (.*)$",
|
|
re.MULTILINE,
|
|
)
|
|
|
|
|
|
class LanguagePython(Language):
|
|
NAME = "python"
|
|
SNIPPET_FILENAME = "python.py"
|
|
SUPPORTS_SYNTAX_CHECK = True
|
|
|
|
@classmethod
|
|
def check_syntax(cls, code: str) -> None:
|
|
ast.parse(code)
|
|
|
|
@classmethod
|
|
def compile(cls, tmpdir: Path, fnames: list[Path]) -> CompileResult:
|
|
trimmed_fnames = trim_commonpath(fnames)
|
|
result = CompileResult()
|
|
|
|
for snippet_fname, trimmed_fname in trimmed_fnames.items():
|
|
target_path = tmpdir / str(trimmed_fname).replace("-", "_")
|
|
target_path.parent.mkdir(parents=True, exist_ok=True)
|
|
dir = tmpdir
|
|
for component in Path(str(trimmed_fname).replace("-", "_")).parts:
|
|
(dir / "__init__.py").touch()
|
|
dir /= component
|
|
copy_template(snippet_fname, target_path)
|
|
result.run_args[snippet_fname] = [
|
|
"uv",
|
|
"--project=templates/python",
|
|
"run",
|
|
target_path,
|
|
]
|
|
|
|
p = subprocess.run(
|
|
[
|
|
"uv",
|
|
"--project=templates/python",
|
|
"run",
|
|
"mypy",
|
|
"--config-file=templates/python/pyproject.toml",
|
|
tmpdir,
|
|
]
|
|
)
|
|
result.has_issues = p.returncode != 0
|
|
|
|
return result
|
|
|
|
@classmethod
|
|
def shorten(cls, contents: str) -> dict[str, str]:
|
|
lines = [
|
|
line
|
|
for line in contents.splitlines(keepends=True)
|
|
if not line.lstrip().startswith("# mypy:")
|
|
]
|
|
return generic_shorten("".join(lines))
|
|
|
|
@classmethod
|
|
def unshorten(cls, contents: str) -> str:
|
|
if "client." in contents and "QdrantClient" not in contents:
|
|
contents = (
|
|
'client = QdrantClient(url="http://localhost:6333") # @hide\n\n'
|
|
+ contents.lstrip()
|
|
)
|
|
|
|
has_imports = set()
|
|
if m := RE_IMPORTS.search(contents):
|
|
has_imports = set(i.strip() for i in m[1].split(","))
|
|
|
|
need_imports = set()
|
|
if "QdrantClient" in contents:
|
|
need_imports.add("QdrantClient")
|
|
if "models." in contents:
|
|
need_imports.add("models")
|
|
|
|
if need_imports - has_imports:
|
|
if m is not None:
|
|
contents = RE_IMPORTS.sub(
|
|
f"from qdrant_client import {', '.join(sorted(need_imports | has_imports))}",
|
|
contents,
|
|
)
|
|
else:
|
|
contents = (
|
|
f"from qdrant_client import {', '.join(sorted(need_imports))} # @hide\n\n"
|
|
+ contents.lstrip()
|
|
)
|
|
|
|
return contents
|