build.db.models.DocumentBaseModel.model_dump() - Code Metrics - Inspection of "Merge pull request #449 from kytos-ng/upgrade/pyth..." - kytos-ng/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( d4a335...2700e4 )

by Vinicius

created 2024-03-28 16:06 UTC

build.db.models.DocumentBaseModel.model_dump() A

↳ Parent: build.db.models

Complexity

Conditions

Size

Total Lines	8
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	5

Importance

Changes

Metric	Value
cc	5
eloc	7
nop	2
dl	0
loc	8
ccs	7
cts	7
cp	1
crap	5
rs	9.3333
c	0
b	0
f	0

"""Models for Mongo DB"""
# pylint: disable=unused-argument,invalid-name,unused-argument
# pylint: disable=no-self-argument,no-name-in-module

from datetime import datetime
from typing import Dict, Literal, Optional, Union

from pydantic import BaseModel, Field, field_validator


class DocumentBaseModel(BaseModel):
    """Base model for Mongo documents"""

    id: str = Field(None, alias="_id")
    inserted_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

    def model_dump(self, **kwargs) -> Dict:
        """Return a dictionary representation of the model"""
        values = super().model_dump(**kwargs)
        if "id" in values and values["id"]:
            values["_id"] = values["id"]
        if "exclude" in kwargs and "_id" in kwargs["exclude"]:
            del values["_id"]
        return values


class CircuitScheduleDoc(BaseModel):
    """EVC circuit schedule model"""

    id: str
    date: Optional[str] = None
    frequency: Optional[str] = None
    interval: Optional[int] = None
    action: str


class TAGDoc(BaseModel):
    """TAG model"""
    tag_type: str
    value: Union[int, str, list[list[int]]]
    mask_list: Optional[list[Union[str, int]]] = None

    @field_validator('value')
    def validate_value(cls, value):
        """Validate value when is a string"""
        if isinstance(value, list):
            return value
        if isinstance(value, int):
            return value
        if isinstance(value, str) and value in ("any", "untagged"):
            return value
        raise ValueError(f"{value} is not allowed as {type(value)}. " +
                         "Allowed strings are 'any' and 'untagged'.")


class UNIDoc(BaseModel):
    """UNI model"""
    tag: Optional[TAGDoc] = None
    interface_id: str


class LinkConstraints(BaseModel):
    """LinkConstraints."""
    bandwidth: Optional[float] = None
    ownership: Optional[str] = None
    reliability: Optional[float] = None
    utilization: Optional[float] = None
    delay: Optional[float] = None
    priority: Optional[int] = None
    not_ownership: Optional[list[str]] = None


class PathConstraints(BaseModel):
    """Pathfinder Constraints."""
    spf_attribute: Optional[Literal["hop", "delay", "priority"]] = None
    spf_max_path_cost: Optional[float] = None
    mandatory_metrics: Optional[LinkConstraints] = None
    flexible_metrics: Optional[LinkConstraints] = None
    minimum_flexible_hits: Optional[int] = None
    undesired_links: Optional[list[str]] = None


class EVCUpdateDoc(DocumentBaseModel):
    """Base model when updating an EVC document"""
    uni_a: Optional[UNIDoc] = None
    uni_z: Optional[UNIDoc] = None
    name: Optional[str] = None
    request_time: Optional[datetime] = None
    start_date: Optional[datetime] = None
    end_date: Optional[datetime] = None
    queue_id: Optional[int] = None
    flow_removed_at: Optional[datetime] = None
    execution_rounds: Optional[int] = None
    bandwidth: Optional[int] = None
    primary_path: Optional[list] = None
    backup_path: Optional[list] = None
    primary_links: Optional[list] = None
    backup_links: Optional[list] = None
    dynamic_backup_path: Optional[bool] = None
    primary_constraints: Optional[PathConstraints] = None
    secondary_constraints: Optional[PathConstraints] = None
    owner: Optional[str] = None
    sb_priority: Optional[int] = None
    service_level: Optional[int] = None
    circuit_scheduler: Optional[list[CircuitScheduleDoc]] = None
    metadata: Optional[dict] = None
    enabled: Optional[bool] = None


class EVCBaseDoc(DocumentBaseModel):
    """Base model for EVC documents"""

    uni_a: UNIDoc
    uni_z: UNIDoc
    name: str
    request_time: Optional[datetime] = None
    start_date: Optional[datetime] = None
    end_date: Optional[datetime] = None
    queue_id: Optional[int] = None
    flow_removed_at: Optional[datetime] = None
    execution_rounds: int = 0
    bandwidth: int = 0
    primary_path: Optional[list] = None
    backup_path: Optional[list] = None
    current_path: Optional[list] = None
    failover_path: Optional[list] = None
    primary_links: Optional[list] = None
    backup_links: Optional[list] = None
    dynamic_backup_path: bool
    primary_constraints: Optional[PathConstraints] = None
    secondary_constraints: Optional[PathConstraints] = None
    creation_time: datetime
    owner: Optional[str] = None
    sb_priority: Optional[int] = None
    service_level: int = 0
    circuit_scheduler: list[CircuitScheduleDoc]
    archived: bool = False
    metadata: Dict = {}
    active: bool
    enabled: bool

    @staticmethod
    def projection() -> Dict:
        """Base projection of EVCBaseDoc model."""
        time_fmt = "%Y-%m-%dT%H:%M:%S"
        return {
            "_id": 0,
            "id": 1,
            "uni_a": 1,
            "uni_z": 1,
            "name": 1,
            "bandwidth": 1,
            "primary_path": 1,
            "backup_path": 1,
            "current_path": 1,
            "failover_path": 1,
            "dynamic_backup_path": 1,
            "sb_priority": {"$ifNull": ["$sb_priority", "$priority", None]},
            "service_level": 1,
            "circuit_scheduler": 1,
            "archived": 1,
            "metadata": 1,
            "active": 1,
            "enabled": 1,
            "execution_rounds": {"$ifNull": ["$execution_rounds", 0]},
            "owner": {"$ifNull": ["$owner", None]},
            "queue_id": {"$ifNull": ["$queue_id", None]},
            "primary_constraints": {"$ifNull": ["$primary_constraints", {}]},
            "secondary_constraints": {"$ifNull": ["$secondary_constraints",
                                      {}]},
            "primary_links": {"$ifNull": ["$primary_links", []]},
            "backup_links": {"$ifNull": ["$backup_links", []]},
            "start_date": {"$dateToString": {
                "format": time_fmt, "date": "$start_date"
            }},
            "creation_time": {"$dateToString": {
                "format": time_fmt, "date": "$creation_time"
            }},
            "request_time": {"$dateToString": {
                "format": time_fmt, "date": {
                    "$ifNull": ["$request_time", "$inserted_at"]
                }
            }},
            "end_date": {"$dateToString": {
                "format": time_fmt, "date": {
                    "$ifNull": ["$end_date", None]
                }
            }},
            "flow_removed_at": {"$dateToString": {
                "format": time_fmt, "date": {
                    "$ifNull": ["$flow_removed_at", None]
                }
            }},
            "updated_at": {"$dateToString": {
                "format": time_fmt, "date": "$updated_at"
            }}
        }


1		"""Models for Mongo DB"""
2		# pylint: disable=unused-argument,invalid-name,unused-argument
3		# pylint: disable=no-self-argument,no-name-in-module
4
5	1	from datetime import datetime
6	1	from typing import Dict, Literal, Optional, Union
7
8	1	from pydantic import BaseModel, Field, field_validator
9
10
11	1	class DocumentBaseModel(BaseModel):
12		"""Base model for Mongo documents"""
13
14	1	id: str = Field(None, alias="_id")
15	1	inserted_at: Optional[datetime] = None
16	1	updated_at: Optional[datetime] = None
17
18	1	def model_dump(self, **kwargs) -> Dict:
19		"""Return a dictionary representation of the model"""
20	1	values = super().model_dump(**kwargs)
21	1	if "id" in values and values["id"]:
22	1	values["_id"] = values["id"]
23	1	if "exclude" in kwargs and "_id" in kwargs["exclude"]:
24	1	del values["_id"]
25	1	return values
26
27
28	1	class CircuitScheduleDoc(BaseModel):
29		"""EVC circuit schedule model"""
30
31	1	id: str
32	1	date: Optional[str] = None
33	1	frequency: Optional[str] = None
34	1	interval: Optional[int] = None
35	1	action: str
36
37
38	1	class TAGDoc(BaseModel):
39		"""TAG model"""
40	1	tag_type: str
41	1	value: Union[int, str, list[list[int]]]
42	1	mask_list: Optional[list[Union[str, int]]] = None
43
44	1	@field_validator('value')
45	1	def validate_value(cls, value):
46		"""Validate value when is a string"""
47	1	if isinstance(value, list):
48	1	return value
49	1	if isinstance(value, int):
50	1	return value
51	1	if isinstance(value, str) and value in ("any", "untagged"):
52	1	return value
53	1	raise ValueError(f"{value} is not allowed as {type(value)}. " +
54		"Allowed strings are 'any' and 'untagged'.")
55
56
57	1	class UNIDoc(BaseModel):
58		"""UNI model"""
59	1	tag: Optional[TAGDoc] = None
60	1	interface_id: str
61
62
63	1	class LinkConstraints(BaseModel):
64		"""LinkConstraints."""
65	1	bandwidth: Optional[float] = None
66	1	ownership: Optional[str] = None
67	1	reliability: Optional[float] = None
68	1	utilization: Optional[float] = None
69	1	delay: Optional[float] = None
70	1	priority: Optional[int] = None
71	1	not_ownership: Optional[list[str]] = None
72
73
74	1	class PathConstraints(BaseModel):
75		"""Pathfinder Constraints."""
76	1	spf_attribute: Optional[Literal["hop", "delay", "priority"]] = None
77	1	spf_max_path_cost: Optional[float] = None
78	1	mandatory_metrics: Optional[LinkConstraints] = None
79	1	flexible_metrics: Optional[LinkConstraints] = None
80	1	minimum_flexible_hits: Optional[int] = None
81	1	undesired_links: Optional[list[str]] = None
82
83
84	1	class EVCUpdateDoc(DocumentBaseModel):
85		"""Base model when updating an EVC document"""
86	1	uni_a: Optional[UNIDoc] = None
87	1	uni_z: Optional[UNIDoc] = None
88	1	name: Optional[str] = None
89	1	request_time: Optional[datetime] = None
90	1	start_date: Optional[datetime] = None
91	1	end_date: Optional[datetime] = None
92	1	queue_id: Optional[int] = None
93	1	flow_removed_at: Optional[datetime] = None
94	1	execution_rounds: Optional[int] = None
95	1	bandwidth: Optional[int] = None
96	1	primary_path: Optional[list] = None
97	1	backup_path: Optional[list] = None
98	1	primary_links: Optional[list] = None
99	1	backup_links: Optional[list] = None
100	1	dynamic_backup_path: Optional[bool] = None
101	1	primary_constraints: Optional[PathConstraints] = None
102	1	secondary_constraints: Optional[PathConstraints] = None
103	1	owner: Optional[str] = None
104	1	sb_priority: Optional[int] = None
105	1	service_level: Optional[int] = None
106	1	circuit_scheduler: Optional[list[CircuitScheduleDoc]] = None
107	1	metadata: Optional[dict] = None
108	1	enabled: Optional[bool] = None
109
110
111	1	class EVCBaseDoc(DocumentBaseModel):
112		"""Base model for EVC documents"""
113
114	1	uni_a: UNIDoc
115	1	uni_z: UNIDoc
116	1	name: str
117	1	request_time: Optional[datetime] = None
118	1	start_date: Optional[datetime] = None
119	1	end_date: Optional[datetime] = None
120	1	queue_id: Optional[int] = None
121	1	flow_removed_at: Optional[datetime] = None
122	1	execution_rounds: int = 0
123	1	bandwidth: int = 0
124	1	primary_path: Optional[list] = None
125	1	backup_path: Optional[list] = None
126	1	current_path: Optional[list] = None
127	1	failover_path: Optional[list] = None
128	1	primary_links: Optional[list] = None
129	1	backup_links: Optional[list] = None
130	1	dynamic_backup_path: bool
131	1	primary_constraints: Optional[PathConstraints] = None
132	1	secondary_constraints: Optional[PathConstraints] = None
133	1	creation_time: datetime
134	1	owner: Optional[str] = None
135	1	sb_priority: Optional[int] = None
136	1	service_level: int = 0
137	1	circuit_scheduler: list[CircuitScheduleDoc]
138	1	archived: bool = False
139	1	metadata: Dict = {}
140	1	active: bool
141	1	enabled: bool
142
143	1	@staticmethod
144	1	def projection() -> Dict:
145		"""Base projection of EVCBaseDoc model."""
146	1	time_fmt = "%Y-%m-%dT%H:%M:%S"
147	1	return {
148		"_id": 0,
149		"id": 1,
150		"uni_a": 1,
151		"uni_z": 1,
152		"name": 1,
153		"bandwidth": 1,
154		"primary_path": 1,
155		"backup_path": 1,
156		"current_path": 1,
157		"failover_path": 1,
158		"dynamic_backup_path": 1,
159		"sb_priority": {"$ifNull": ["$sb_priority", "$priority", None]},
160		"service_level": 1,
161		"circuit_scheduler": 1,
162		"archived": 1,
163		"metadata": 1,
164		"active": 1,
165		"enabled": 1,
166		"execution_rounds": {"$ifNull": ["$execution_rounds", 0]},
167		"owner": {"$ifNull": ["$owner", None]},
168		"queue_id": {"$ifNull": ["$queue_id", None]},
169		"primary_constraints": {"$ifNull": ["$primary_constraints", {}]},
170		"secondary_constraints": {"$ifNull": ["$secondary_constraints",
171		{}]},
172		"primary_links": {"$ifNull": ["$primary_links", []]},
173		"backup_links": {"$ifNull": ["$backup_links", []]},
174		"start_date": {"$dateToString": {
175		"format": time_fmt, "date": "$start_date"
176		}},
177		"creation_time": {"$dateToString": {
178		"format": time_fmt, "date": "$creation_time"
179		}},
180		"request_time": {"$dateToString": {
181		"format": time_fmt, "date": {
182		"$ifNull": ["$request_time", "$inserted_at"]
183		}
184		}},
185		"end_date": {"$dateToString": {
186		"format": time_fmt, "date": {
187		"$ifNull": ["$end_date", None]
188		}
189		}},
190		"flow_removed_at": {"$dateToString": {
191		"format": time_fmt, "date": {
192		"$ifNull": ["$flow_removed_at", None]
193		}
194		}},
195		"updated_at": {"$dateToString": {
196		"format": time_fmt, "date": "$updated_at"
197		}}
198		}
199

kytos-ng / mef_eline

Push — master ( d4a335...2700e4 )

build.db.models.DocumentBaseModel.model_dump() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like