Coverage for functions \ flipdare \ error \ data_load_error.py: 93%
44 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
1#!/usr/bin/env python
2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved.
3#
4# This file is part of Flipdare's proprietary software and contains
5# confidential and copyrighted material. Unauthorised copying,
6# modification, distribution, or use of this file is strictly
7# prohibited without prior written permission from Flipdare Pty Ltd.
8#
9# This software includes third-party components licensed under MIT,
10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.
11#
14from typing import Self
16from pydantic import ValidationError
17from flipdare.error.app_error import AppError
18from flipdare.generated import AppErrorCode
19from flipdare.generated.schema.error.error_code_schema import ErrorCodeSchema
20from flipdare.message.error_message import ErrorMessage
22__all__ = ["DataLoadError"]
25class DataLoadError(AppError):
26 CODE = AppErrorCode.DATA_LOAD
28 @classmethod
29 def model(
30 cls,
31 class_name: str,
32 missing_code: str,
33 error: ValidationError | list[str],
34 ) -> Self:
35 return cls._create(source=class_name, missing_code=missing_code, error=error)
37 @classmethod
38 def malformed(
39 cls,
40 endpoint: str,
41 missing_code: str,
42 error: ValidationError | list[str],
43 ) -> Self:
44 return cls._create(source=endpoint, missing_code=missing_code, error=error)
46 @classmethod
47 def _create(
48 cls,
49 source: str,
50 missing_code: str,
51 error: ValidationError | list[str],
52 ) -> Self:
53 missing = error if isinstance(error, list) else cls.parse_error(error)
54 missing_str = "Missing Fields: " + ", ".join(missing)
56 return cls(
57 source=source,
58 message=ErrorMessage.DATA_LOAD_ERROR.formatted(ErrorCodeSchema(code=missing_code)),
59 error_code=cls.CODE,
60 cause=missing_str,
61 error=error if isinstance(error, Exception) else None,
62 )
64 @staticmethod
65 def parse_error(error: ValidationError) -> list[str]:
66 messages = []
67 for err in error.errors():
68 error_type = err["type"]
69 loc = err["loc"]
71 msg: str | None = None
72 if "dict_type" in error_type:
73 # this is a problem with the entire dict, so we won't have a field name. Just report the error type
74 msg = "Invalid Request"
75 else:
76 # get the field name ..
77 field_name = "unknown"
78 if loc and len(loc) > 0 and isinstance(loc[0], str):
79 field_name = loc[0]
81 reason: str
82 if "parsing" in error_type:
83 reason = "Invalid"
84 elif "missing" in error_type:
85 reason = "Missing"
86 else:
87 reason = "Unknown"
89 if field_name == "unknown" and reason == "Unknown":
90 msg = "Screwy Input"
91 else:
92 msg = f"{field_name} ({reason})"
94 messages.append(msg)
96 if len(messages) == 0:
97 messages.append("Unknown Error")
99 return messages