8766121206

Committed 20 Apr 2024 03:45PM UTC coverage: 83.956% (-2.9%) from 86.892%

Build # 8766121206

Build Type

push

github

Committed by

andgineer

Commit Message

upgrade reqs

Run Details

2 of 2 new or added lines in 2 files covered. (100.0%)

89 existing lines in 23 files now uncovered.

1057 of 1259 relevant lines covered (83.96%)

0.84 hits per line

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

91.23

/src/controllers/models.py

import enum
import random
from typing import Any, Dict, Optional, Tuple, Type, Union

from schematics.exceptions import ConversionError
from schematics.models import Model
from schematics.types import BaseType, IntType, ListType, StringType

PAGE_DEFAULT = 1
PER_PAGE_DEFAULT = 30

ApiResult = Union[Dict[str, Any], Tuple[str, int]]


class HttpCode:
    """HTTP result codes."""

    success = 200  # Default HTTP result code
    successes = [success]
    unhandled_exception = 500  # Unhandled exception
    unauthorized = 403
    wrong_request = 501  # Wrong request format etc
    logic_error = 400  # Application level error like user already exists and so on
    no_token = 401  # Request without token for operation that requires one


@enum.unique
class UserGroup(enum.Enum):
    """User group."""

    FULL = "full"
    ADMIN = "admin"
    GUEST = "guest"


class APIBaseError(Exception):
    """Error."""

    status: Optional[int] = (
        None  # non-abstract descendants should replace that with specific HTTP error
    )


class APIError(APIBaseError):
    """Error."""

    status = HttpCode.wrong_request


class APIUnauthError(APIBaseError):
    """Error."""

    status = HttpCode.unauthorized


class APILogicError(APIBaseError):
    """Error."""

    status = HttpCode.logic_error


class APIValueError(APIError):
    """Error."""


class APPNoTokenError(APIBaseError):
    """Error."""


class Paging(Model):  # type: ignore
    """Validation of list's requests page/per_page parameters."""

    page = IntType(min_value=1, default=PAGE_DEFAULT)
    per_page = IntType(min_value=1, default=PER_PAGE_DEFAULT)

    def __init__(self, raw_data: Dict[str, Any], **kwargs: Any) -> None:
        """Init."""
        if "page" in raw_data and not str(raw_data["page"]).strip():
            del raw_data["page"]
        if "per_page" in raw_data and not str(raw_data["per_page"]).strip():
            del raw_data["per_page"]
        super().__init__(raw_data, **kwargs)


class APIModel(Model):  # type: ignore
    """Model for API responses."""

    @property
    def as_dict(self) -> Dict[str, Any]:
        """Export with conversion of all fields into safe for json string."""
        return self.to_primitive()  # type: ignore

    @property
    def to_orm(self) -> Dict[str, Any]:
        """Export for ORM (SQLAlchemy).

        Dictionary without convertion of dates
        """
        result: Dict[str, Any] = self.to_native()

        empty_fields = {field for field in result if result[field] is None}
        for field in empty_fields:
            del result[field]
        return result

    def from_dict(self, data: Dict[str, Any]) -> "APIModel":
        """Import from dict."""
        super().__init__({param: data[param] for param in self.keys()})
        return self

    def from_orm(self, data: Dict[str, Any]) -> "APIModel":
        """Import from ORM (SQLAlchemy)."""
        super().__init__({param: getattr(data, param) for param in self.keys()})
        return self


class EnumType(BaseType):  # type: ignore
    """Converts Python Enum into the string."""

    primitive_type = str
    native_type = enum.Enum

    MESSAGES = {
        "convert": ("Couldn't interpret '{0}' value as Enum."),
        "find": "Couldnt find {value} in {choices}",
    }

    def __init__(self, enum: Optional[Type[enum.Enum]] = None, **kwargs: Any) -> None:
        """Init."""
        self.enum = enum
        super().__init__(**kwargs)

    def _mock(self, context: Any = None) -> str:  # pylint: disable=unused-argument
        """Return random enum value."""
        assert self.enum is not None
        return random.choice(list(self.enum.__members__))

    def to_native(
        self,
        value: Union[enum.Enum, str],
        context: Any = None,  # pylint: disable=unused-argument
    ) -> enum.Enum:
        """Convert to native.

        If `value` is instance of `EnumType` return it as it is.
        In other case return `EnumType` instance with name == `value` or raise exception if no such instance.
        """
        assert self.enum is not None
        if isinstance(value, self.enum):
            return value
        if not isinstance(value, str):
            raise ConversionError(self.messages["convert"].format(value))
        assert self.enum is not None
        try:
            for member in self.enum.__members__:
                if member.lower() == value.lower():
                    return self.enum.__members__[member]
            raise ValueError(
                self.messages["find"].format(choices=self.enum.__members__, value=value)
            )
        except (ValueError, TypeError) as e:
            raise ConversionError(self.messages["convert"].format(value)) from e

    def to_primitive(
        self,
        value: enum.Enum,
        context: Any = None,  # pylint: disable=unused-argument
    ) -> Union[str, int]:
        """Convert to primitive."""
        return value.value  # type: ignore


class TokenReply(APIModel):
    """Get_token reply."""

    token = StringType(required=True)


class UserShort(APIModel):
    """User in list."""

    id = StringType()
    group = EnumType(enum=UserGroup)
    email = StringType()
    name = StringType()


class UpdateUser(APIModel):
    """Update user."""

    group = EnumType(enum=UserGroup)
    email = StringType(required=True)
    password = StringType()
    name = StringType()


class NewUser(APIModel):
    """Extend validation.

    https://schematics.readthedocs.io/en/latest/usage/validation.html#extending-validation
    """

    group = EnumType(enum=UserGroup, required=True)
    email = StringType(required=True)
    password = StringType(required=True)
    name = StringType()


class NewUserReply(APIModel):
    """Create result."""

    id = IntType(required=True)


class User(APIModel):
    """User."""

    id = StringType()
    group = EnumType(enum=UserGroup)
    email = StringType()
    name = StringType(default="")


class UserCredentials(APIModel):
    """User credentials."""

    email = StringType()
    password = StringType()


class UsersList(APIModel):
    """List of users."""

    data = ListType(StringType, required=True)
    total = IntType(required=True)

1	import enum	1✔
2	import random	1✔
3	from typing import Any, Dict, Optional, Tuple, Type, Union	1✔
4
5	from schematics.exceptions import ConversionError	1✔
6	from schematics.models import Model	1✔
7	from schematics.types import BaseType, IntType, ListType, StringType	1✔
8
9	PAGE_DEFAULT = 1	1✔
10	PER_PAGE_DEFAULT = 30	1✔
11
12	ApiResult = Union[Dict[str, Any], Tuple[str, int]]	1✔
13
14
15	class HttpCode:	1✔
16	"""HTTP result codes."""
17
18	success = 200 # Default HTTP result code	1✔
19	successes = [success]	1✔
20	unhandled_exception = 500 # Unhandled exception	1✔
21	unauthorized = 403	1✔
22	wrong_request = 501 # Wrong request format etc	1✔
23	logic_error = 400 # Application level error like user already exists and so on	1✔
24	no_token = 401 # Request without token for operation that requires one	1✔
25
26
27	@enum.unique	1✔
28	class UserGroup(enum.Enum):	1✔
29	"""User group."""
30
31	FULL = "full"	1✔
32	ADMIN = "admin"	1✔
33	GUEST = "guest"	1✔
34
35
36	class APIBaseError(Exception):	1✔
37	"""Error."""
38
39	status: Optional[int] = (	1✔
40	None # non-abstract descendants should replace that with specific HTTP error
41	)
42
43
44	class APIError(APIBaseError):	1✔
45	"""Error."""
46
47	status = HttpCode.wrong_request	1✔
48
49
50	class APIUnauthError(APIBaseError):	1✔
51	"""Error."""
52
53	status = HttpCode.unauthorized	1✔
54
55
56	class APILogicError(APIBaseError):	1✔
57	"""Error."""
58
59	status = HttpCode.logic_error	1✔
60
61
62	class APIValueError(APIError):	1✔
63	"""Error."""
64
65
66	class APPNoTokenError(APIBaseError):	1✔
67	"""Error."""
68
69
70	class Paging(Model): # type: ignore	1✔
71	"""Validation of list's requests page/per_page parameters."""
72
73	page = IntType(min_value=1, default=PAGE_DEFAULT)	1✔
74	per_page = IntType(min_value=1, default=PER_PAGE_DEFAULT)	1✔
75
76	def __init__(self, raw_data: Dict[str, Any], **kwargs: Any) -> None:	1✔
77	"""Init."""
78	if "page" in raw_data and not str(raw_data["page"]).strip():	1✔
UNCOV 79	del raw_data["page"]	×
80	if "per_page" in raw_data and not str(raw_data["per_page"]).strip():	1✔
UNCOV 81	del raw_data["per_page"]	×
82	super().__init__(raw_data, **kwargs)	1✔
83
84
85	class APIModel(Model): # type: ignore	1✔
86	"""Model for API responses."""
87
88	@property	1✔
89	def as_dict(self) -> Dict[str, Any]:	1✔
90	"""Export with conversion of all fields into safe for json string."""
91	return self.to_primitive() # type: ignore	1✔
92
93	@property	1✔
94	def to_orm(self) -> Dict[str, Any]:	1✔
95	"""Export for ORM (SQLAlchemy).
96
97	Dictionary without convertion of dates
98	"""
99	result: Dict[str, Any] = self.to_native()	1✔
100
101	empty_fields = {field for field in result if result[field] is None}	1✔
102	for field in empty_fields:	1✔
103	del result[field]	1✔
104	return result	1✔
105
106	def from_dict(self, data: Dict[str, Any]) -> "APIModel":	1✔
107	"""Import from dict."""
UNCOV 108	super().__init__({param: data[param] for param in self.keys()})	×
UNCOV 109	return self	×
110
111	def from_orm(self, data: Dict[str, Any]) -> "APIModel":	1✔
112	"""Import from ORM (SQLAlchemy)."""
113	super().__init__({param: getattr(data, param) for param in self.keys()})	1✔
114	return self	1✔
115
116
117	class EnumType(BaseType): # type: ignore	1✔
118	"""Converts Python Enum into the string."""
119
120	primitive_type = str	1✔
121	native_type = enum.Enum	1✔
122
123	MESSAGES = {	1✔
124	"convert": ("Couldn't interpret '{0}' value as Enum."),
125	"find": "Couldnt find {value} in {choices}",
126	}
127
128	def __init__(self, enum: Optional[Type[enum.Enum]] = None, **kwargs: Any) -> None:	1✔
129	"""Init."""
130	self.enum = enum	1✔
131	super().__init__(**kwargs)	1✔
132
133	def _mock(self, context: Any = None) -> str: # pylint: disable=unused-argument	1✔
134	"""Return random enum value."""
UNCOV 135	assert self.enum is not None	×
UNCOV 136	return random.choice(list(self.enum.__members__))	×
137
138	def to_native(	1✔
139	self,
140	value: Union[enum.Enum, str],
141	context: Any = None, # pylint: disable=unused-argument
142	) -> enum.Enum:
143	"""Convert to native.
144
145	If `value` is instance of `EnumType` return it as it is.
146	In other case return `EnumType` instance with name == `value` or raise exception if no such instance.
147	"""
148	assert self.enum is not None	1✔
149	if isinstance(value, self.enum):	1✔
150	return value	1✔
151	if not isinstance(value, str):	1✔
UNCOV 152	raise ConversionError(self.messages["convert"].format(value))	×
153	assert self.enum is not None	1✔
154	try:	1✔
155	for member in self.enum.__members__:	1✔
156	if member.lower() == value.lower():	1✔
157	return self.enum.__members__[member]	1✔
UNCOV 158	raise ValueError(	×
159	self.messages["find"].format(choices=self.enum.__members__, value=value)
160	)
UNCOV 161	except (ValueError, TypeError) as e:	×
UNCOV 162	raise ConversionError(self.messages["convert"].format(value)) from e	×
163
164	def to_primitive(	1✔
165	self,
166	value: enum.Enum,
167	context: Any = None, # pylint: disable=unused-argument
168	) -> Union[str, int]:
169	"""Convert to primitive."""
170	return value.value # type: ignore	1✔
171
172
173	class TokenReply(APIModel):	1✔
174	"""Get_token reply."""
175
176	token = StringType(required=True)	1✔
177
178
179	class UserShort(APIModel):	1✔
180	"""User in list."""
181
182	id = StringType()	1✔
183	group = EnumType(enum=UserGroup)	1✔
184	email = StringType()	1✔
185	name = StringType()	1✔
186
187
188	class UpdateUser(APIModel):	1✔
189	"""Update user."""
190
191	group = EnumType(enum=UserGroup)	1✔
192	email = StringType(required=True)	1✔
193	password = StringType()	1✔
194	name = StringType()	1✔
195
196
197	class NewUser(APIModel):	1✔
198	"""Extend validation.
199
200	https://schematics.readthedocs.io/en/latest/usage/validation.html#extending-validation
201	"""
202
203	group = EnumType(enum=UserGroup, required=True)	1✔
204	email = StringType(required=True)	1✔
205	password = StringType(required=True)	1✔
206	name = StringType()	1✔
207
208
209	class NewUserReply(APIModel):	1✔
210	"""Create result."""
211
212	id = IntType(required=True)	1✔
213
214
215	class User(APIModel):	1✔
216	"""User."""
217
218	id = StringType()	1✔
219	group = EnumType(enum=UserGroup)	1✔
220	email = StringType()	1✔
221	name = StringType(default="")	1✔
222
223
224	class UserCredentials(APIModel):	1✔
225	"""User credentials."""
226
227	email = StringType()	1✔
228	password = StringType()	1✔
229
230
231	class UsersList(APIModel):	1✔
232	"""List of users."""
233
234	data = ListType(StringType, required=True)	1✔
235	total = IntType(required=True)	1✔

andgineer / api-db-prototype / 8766121206

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