mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-03 07:23:33 +00:00
0faaeccfb6
The narrowed design no longer changes the daily tag spend unique key, and main's schema compatibility check already passes this schema, so its script and tests go back to main's version. Adds a test that a batch without a team never clears a stored team and one that the spend helper hands the virtual key's team to the tag rows, and drops a comment
77 lines
2.8 KiB
Python
77 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from collections.abc import Iterator, Mapping
|
|
from pathlib import Path
|
|
from types import MappingProxyType
|
|
from typing import Final
|
|
|
|
from pydantic import TypeAdapter
|
|
|
|
Schema = Mapping[str, Mapping[str, str]]
|
|
MODELS: Final = re.compile(r"^[ \t]*model\s+(\w+)\s*\{(.*?)^[ \t]*\}", re.MULTILINE | re.DOTALL)
|
|
|
|
|
|
def parse_model_fields(schema: str) -> Schema:
|
|
def fields(body: str) -> Mapping[str, str]:
|
|
parts: Final = tuple(line.split() for line in body.splitlines())
|
|
return MappingProxyType({part[0]: " ".join(part[1:]) for part in parts if len(part) >= 2})
|
|
|
|
return MappingProxyType({match[1]: fields(match[2]) for match in MODELS.finditer(schema)})
|
|
|
|
|
|
def _field_changes(model: str, field: str, old_type: str, new_fields: Mapping[str, str]) -> Iterator[str]:
|
|
if field not in new_fields:
|
|
yield f"Breaking: Field {model}.{field} was removed"
|
|
return
|
|
new_type: Final = new_fields[field]
|
|
if old_type == new_type:
|
|
return
|
|
if "?" in old_type and "?" not in new_type:
|
|
yield f"Breaking: Field {model}.{field} changed from optional to required"
|
|
if not old_type.startswith(new_type.split("?")[0]):
|
|
yield f"Breaking: Field {model}.{field} changed type from {old_type} to {new_type}"
|
|
|
|
|
|
def _model_changes(model: str, old_fields: Mapping[str, str], new_schema: Schema) -> Iterator[str]:
|
|
if model not in new_schema:
|
|
yield f"Breaking: Model {model} was removed"
|
|
return
|
|
for field, old_type in old_fields.items():
|
|
yield from _field_changes(model, field, old_type, new_schema[model])
|
|
|
|
|
|
def check_breaking_changes(old_schema: Schema, new_schema: Schema) -> tuple[str, ...]:
|
|
def changes() -> Iterator[str]:
|
|
for model, fields in old_schema.items():
|
|
yield from _model_changes(model, fields, new_schema)
|
|
|
|
return tuple(changes())
|
|
|
|
|
|
def main() -> int:
|
|
parser: Final = argparse.ArgumentParser()
|
|
parser.add_argument("--base-ref", required=True)
|
|
options: Final = parser.parse_args()
|
|
base_ref: Final = TypeAdapter(str).validate_python(options.base_ref)
|
|
base: Final = subprocess.run(
|
|
("git", "merge-base", "HEAD", base_ref), capture_output=True, text=True, check=True
|
|
).stdout.strip()
|
|
previous: Final = subprocess.run(
|
|
("git", "show", f"{base}:schema.prisma"), capture_output=True, text=True, check=True
|
|
).stdout
|
|
changes: Final = check_breaking_changes(
|
|
parse_model_fields(previous), parse_model_fields(Path("schema.prisma").read_text())
|
|
)
|
|
if changes:
|
|
sys.stderr.write("\n".join(changes) + "\n")
|
|
return 1
|
|
sys.stdout.write(f"Schema compatibility passed against merge base {base}\n")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|