17220817776

Committed 25 Aug 2025 09:09PM UTC coverage: 80.753% (+1.7%) from 79.016%

Build # 17220817776

Build Type

Pull #96

github

Committed by

actions-user

Commit Message

Apply automatic documentation changes

Pull Request Pull Request #96: Feat/add additional-data column

Run Details

80 of 85 new or added lines in 5 files covered. (94.12%)

5 existing lines in 2 files now uncovered.

558 of 691 relevant lines covered (80.75%)

0.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.45

/src/audit/routes/system.py

import json, hashlib
from pydantic import BaseModel
from typing import Type
from fastapi import APIRouter, FastAPI, Request, Depends
from typing import Union, get_args, get_origin

from ..db import DataAccessLayer, get_data_access_layer

from ..models import CreateLoginLogInput, CreatePresignedUrlLogInput

router = APIRouter()

# BE SURE TO UPDATE BOTH "version" AND "fingerprints" on schema/model change.
CURRENT_SCHEMA_VERSIONS = {
    "login": {
        "version": 2.1,
        "fingerprint": "77b3057a8451ea35f0454dfec6e5c39e107c444c9e17b63e12d8a3a4fe1e53e9",  # pragma: allowlist-secret
    },
    "presigned_url": {
        "version": 1.1,
        "fingerprint": "9acc1cbab580d3d1e073a49e80d62896a2509d4f9f2f1c450d6bbf2b5779c578",  # pragma: allowlist-secret
    },
}


def _model_fingerprint(model_cls: Type[BaseModel]) -> str:
    """
    Takes a pydantic class model and returns a hash of the model's schema
    """
    schema_dict = model_cls.model_json_schema()
    schema_utf8 = json.dumps(schema_dict, sort_keys=True).encode(encoding="utf-8")
    return hashlib.sha256(schema_utf8).hexdigest()


def _pretty_type(t) -> str:
    """
    Human-readable short type name. Includes a question mark for optional fields.

    str                -> "str"
    list[str]          -> "list"
    Union[str, None]   -> "str?"
    """
    origin = get_origin(t)

    # handle Optional types (Union types with None)
    if origin is Union and type(None) in get_args(t):
        non_none_args = [a for a in get_args(t) if a is not type(None)]
        if not non_none_args:
            raise TypeError(
                "Optional type must include at least one non-None type, e.g. Union[str, None]"
            )
        base = non_none_args[0]
        return f"{getattr(base, '__name__', str(base))}?"

    # handle parameterized generic type
    if origin is not None:
        return getattr(origin, "__name__", str(origin))

    # handle regular unparameterized types
    return getattr(t, "__name__", str(t))


def _get_pydantic_model(model_class) -> dict[str, str]:
    """
    Return {"field_name": "type"} for every field in a Pydantic model
    """
    raw_pydantic_fields = getattr(model_class, "model_fields", {})

    model = {}
    for name, field in raw_pydantic_fields.items():
        annotation = getattr(field, "annotation", None)
        model[name] = _pretty_type(annotation)
    return model


@router.get("/_schema")
def get_schema() -> dict:
    """
    GET audit service schema model versions and details.
    404s should assume legacy model.
    """
    return {
        "login": {
            "version": CURRENT_SCHEMA_VERSIONS["login"]["version"],
            "model": _get_pydantic_model(CreateLoginLogInput),
        },
        "presigned_url": {
            "version": CURRENT_SCHEMA_VERSIONS["presigned_url"]["version"],
            "model": _get_pydantic_model(CreatePresignedUrlLogInput),
        },
    }


@router.get("/_version")
def get_version(request: Request) -> dict:
    return dict(version=request.app.version)


@router.get("/")
@router.get("/_status")
async def get_status(
    data_access_layer: DataAccessLayer = Depends(get_data_access_layer),
) -> dict:
    await data_access_layer.test_connection()
    return dict(status="OK")


def init_app(app: FastAPI) -> None:
    app.include_router(router, tags=["System"])

1	import json, hashlib	1✔
2	from pydantic import BaseModel	1✔
3	from typing import Type	1✔
4	from fastapi import APIRouter, FastAPI, Request, Depends	1✔
5	from typing import Union, get_args, get_origin	1✔
6
7	from ..db import DataAccessLayer, get_data_access_layer	1✔
8
9	from ..models import CreateLoginLogInput, CreatePresignedUrlLogInput	1✔
10
11	router = APIRouter()	1✔
12
13	# BE SURE TO UPDATE BOTH "version" AND "fingerprints" on schema/model change.
14	CURRENT_SCHEMA_VERSIONS = {	1✔
15	"login": {
16	"version": 2.1,
17	"fingerprint": "77b3057a8451ea35f0454dfec6e5c39e107c444c9e17b63e12d8a3a4fe1e53e9", # pragma: allowlist-secret
18	},
19	"presigned_url": {
20	"version": 1.1,
21	"fingerprint": "9acc1cbab580d3d1e073a49e80d62896a2509d4f9f2f1c450d6bbf2b5779c578", # pragma: allowlist-secret
22	},
23	}
24
25
26	def _model_fingerprint(model_cls: Type[BaseModel]) -> str:	1✔
27	"""
28	Takes a pydantic class model and returns a hash of the model's schema
29	"""
30	schema_dict = model_cls.model_json_schema()	1✔
31	schema_utf8 = json.dumps(schema_dict, sort_keys=True).encode(encoding="utf-8")	1✔
32	return hashlib.sha256(schema_utf8).hexdigest()	1✔
33
34
35	def _pretty_type(t) -> str:	1✔
36	"""
37	Human-readable short type name. Includes a question mark for optional fields.
38
39	str -> "str"
40	list[str] -> "list"
41	Union[str, None] -> "str?"
42	"""
43	origin = get_origin(t)	1✔
44
45	# handle Optional types (Union types with None)
46	if origin is Union and type(None) in get_args(t):	1✔
47	non_none_args = [a for a in get_args(t) if a is not type(None)]	1✔
48	if not non_none_args:	1✔
NEW UNCOV 49	raise TypeError(	×
50	"Optional type must include at least one non-None type, e.g. Union[str, None]"
51	)
52	base = non_none_args[0]	1✔
53	return f"{getattr(base, '__name__', str(base))}?"	1✔
54
55	# handle parameterized generic type
56	if origin is not None:	1✔
57	return getattr(origin, "__name__", str(origin))	1✔
58
59	# handle regular unparameterized types
60	return getattr(t, "__name__", str(t))	1✔
61
62
63	def _get_pydantic_model(model_class) -> dict[str, str]:	1✔
64	"""
65	Return {"field_name": "type"} for every field in a Pydantic model
66	"""
67	raw_pydantic_fields = getattr(model_class, "model_fields", {})	1✔
68
69	model = {}	1✔
70	for name, field in raw_pydantic_fields.items():	1✔
71	annotation = getattr(field, "annotation", None)	1✔
72	model[name] = _pretty_type(annotation)	1✔
73	return model	1✔
74
75
76	@router.get("/_schema")	1✔
77	def get_schema() -> dict:	1✔
78	"""
79	GET audit service schema model versions and details.
80	404s should assume legacy model.
81	"""
82	return {	1✔
83	"login": {
84	"version": CURRENT_SCHEMA_VERSIONS["login"]["version"],
85	"model": _get_pydantic_model(CreateLoginLogInput),
86	},
87	"presigned_url": {
88	"version": CURRENT_SCHEMA_VERSIONS["presigned_url"]["version"],
89	"model": _get_pydantic_model(CreatePresignedUrlLogInput),
90	},
91	}
92
93
94	@router.get("/_version")	1✔
95	def get_version(request: Request) -> dict:	1✔
96	return dict(version=request.app.version)	1✔
97
98
99	@router.get("/")	1✔
100	@router.get("/_status")	1✔
101	async def get_status(	1✔
102	data_access_layer: DataAccessLayer = Depends(get_data_access_layer),
103	) -> dict:
104	await data_access_layer.test_connection()	1✔
UNCOV 105	return dict(status="OK")	×
106
107
108	def init_app(app: FastAPI) -> None:	1✔
109	app.include_router(router, tags=["System"])	1✔

uc-cdis / audit-service / 17220817776

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous