ragit 0.4.5

git-like rag pipeline
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

import os
import re
import subprocess
from typing import Optional
from utils import goto_root

def cargo_tests():
    goto_root()
    errors = run_cargo_test(
        location="core",
        additional_actions=[
            ["cargo", "test", "--release", "--features=csv,korean,pdf,svg"],
            ["cargo", "doc", "--features=csv,korean,pdf,svg"],
        ],
    )

    for crate in [
        "crates/api",
        "crates/cli",
        "crates/fs",
        "crates/ignore",
        "crates/korean",
        "crates/pdl",
        "ragithub/frontend",
        "ragithub/backend",
    ]:
        os.chdir(crate)
        errors += run_cargo_test(
            location=crate,

            # build artifacts take more than 20GiB
            additional_actions=[["cargo", "clean"]],
        )
        os.chdir("../..")

    if len(errors) > 0:
        raise Exception("\n\n".join(errors))

# returns error messages, if exists
def run_cargo_test(
    location: str,
    additional_actions: Optional[list[list[str]]] = None,
) -> list[str]:
    errors = []
    actions = [
        ["cargo", "test"],
        ["cargo", "test", "--release"],
        ["cargo", "doc"],
    ]
    actions += additional_actions or []

    for action in actions:
        print(f"running `{' '.join(action)}` at `{location}`")
        result = subprocess.run(action, capture_output=True, text=True)

        if result.returncode != 0 or has_warning(result.stderr):
            errors.append(f"""
#####################
### path: command ###
{os.getcwd()}: {' '.join(action)}

### status_code ###
{result.returncode}

### stdout ###
{result.stdout}

### stderr ###
{clean_cargo_output(result.stderr)}
""")

    return errors

def has_warning(stderr: str) -> bool:
    warnings = re.search(r"warning\:.+generated\s(\d+)\swarning", stderr)
    return warnings is not None and int(warnings.group(1)) > 0

def clean_cargo_output(stdout: str) -> str:
    i_dont_want_to_see_these = lambda line: re.match(r"^\s*Compiling\s.+v\d.+", line) or\
        re.match(r"^\s*Documenting\s.+v\d.+", line) or\
        re.match(r"^\s*Checking\s.+v\d.+", line)

    return "\n".join([line for line in stdout.split("\n") if not i_dont_want_to_see_these(line)])