Skip to content

Commit

Permalink
More types!
Browse files Browse the repository at this point in the history
  • Loading branch information
mamiksik committed Apr 17, 2020
1 parent 06320bf commit f160c93
Show file tree
Hide file tree
Showing 2 changed files with 24 additions and 20 deletions.
8 changes: 4 additions & 4 deletions pyodata/model/elements.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import logging
from abc import abstractmethod
from enum import Enum
from typing import Union
from typing import Union, Dict

from pyodata.policies import ParserError
from pyodata.config import Config
Expand Down Expand Up @@ -104,7 +104,7 @@ def __getattr__(self, item):


class Identifier:
def __init__(self, name):
def __init__(self, name: str):
super(Identifier, self).__init__()

self._name = name
Expand Down Expand Up @@ -638,7 +638,7 @@ def __init__(self, name, label, is_value_list):
self._label = label
self._is_value_list = is_value_list
self._key = list()
self._properties = dict()
self._properties: Dict[str, 'StructTypeProperty'] = dict()

@property
def label(self):
Expand All @@ -648,7 +648,7 @@ def label(self):
def is_value_list(self):
return self._is_value_list

def proprty(self, property_name):
def proprty(self, property_name: str) -> 'StructTypeProperty':
try:
return self._properties[property_name]
except KeyError:
Expand Down
36 changes: 20 additions & 16 deletions pyodata/v2/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -793,7 +793,11 @@ def __init__(self, service: 'Service', entity_set: Union[EntitySet, 'EntitySetPr
pass

def __repr__(self) -> str:
return self._entity_key.to_key_string()
entity_key = self._entity_key
if entity_key is None:
raise PyODataException('Entity key is None')

return entity_key.to_key_string()

def __getattr__(self, attr: str) -> Any:
try:
Expand Down Expand Up @@ -844,8 +848,7 @@ def nav(self, nav_property: str) -> Union['NavEntityProxy', 'EntitySetProxy']:

def get_path(self) -> str:
"""Returns this entity's relative path - e.g. EntitySet(KEY)"""

return self._entity_set._name + self._entity_key.to_key_string() # pylint: disable=protected-access
return str(self._entity_set._name + self._entity_key.to_key_string()) # pylint: disable=protected-access

def get_proprty(self, name: str, connection: Optional[requests.Session] = None) -> ODataHttpRequest:
"""Returns value of the property"""
Expand Down Expand Up @@ -886,7 +889,7 @@ def value_get_handler(key: Any, response: requests.Response) -> requests.Respons
connection=connection)

@property
def entity_set(self) -> EntitySet:
def entity_set(self) -> Optional[Union['EntitySet', 'EntitySetProxy']]:
"""Entity set related to this entity"""

return self._entity_set
Expand All @@ -906,7 +909,7 @@ def url(self) -> str:

return urljoin(service_url, entity_path)

def equals(self, other) -> bool:
def equals(self, other: 'EntityProxy') -> bool:
"""Returns true if the self and the other contains the same data"""
# pylint: disable=W0212
return self._cache == other._cache
Expand Down Expand Up @@ -1057,7 +1060,8 @@ def _get_nav_entity(self, master_key: EntityKey, nav_property: str,
navigation_entity_set: EntitySet) -> NavEntityGetRequest:
"""Get entity based on provided key of the master and Navigation property name"""

def get_entity_handler(parent, nav_property, navigation_entity_set, response) -> NavEntityProxy:
def get_entity_handler(parent: EntityProxy, nav_property: str, navigation_entity_set: EntitySet,
response: requests.Response) -> NavEntityProxy:
"""Gets entity from HTTP response"""

if response.status_code != requests.codes.ok:
Expand All @@ -1082,10 +1086,10 @@ def get_entity_handler(parent, nav_property, navigation_entity_set, response) ->
self,
nav_property)

def get_entity(self, key=None, **args):
def get_entity(self, key=None, **args) -> EntityGetRequest:
"""Get entity based on provided key properties"""

def get_entity_handler(response):
def get_entity_handler(response: requests.Response) -> EntityProxy:
"""Gets entity from HTTP response"""

if response.status_code != requests.codes.ok:
Expand All @@ -1108,7 +1112,7 @@ def get_entity_handler(response):
def get_entities(self):
"""Get all entities"""

def get_entities_handler(response):
def get_entities_handler(response: requests.Response) -> Union[List[EntityProxy], int]:
"""Gets entity set from HTTP Response"""

if response.status_code != requests.codes.ok:
Expand All @@ -1133,10 +1137,10 @@ def get_entities_handler(response):
return GetEntitySetRequest(self._service.url, self._service.connection, get_entities_handler,
self._parent_last_segment + entity_set_name, self._entity_set.entity_type)

def create_entity(self, return_code=requests.codes.created):
def create_entity(self, return_code: int = requests.codes.created) -> EntityCreateRequest:
"""Creates a new entity in the given entity-set."""

def create_entity_handler(response):
def create_entity_handler(response: requests.Response) -> EntityProxy:
"""Gets newly created entity encoded in HTTP Response"""

if response.status_code != return_code:
Expand All @@ -1150,10 +1154,10 @@ def create_entity_handler(response):
return EntityCreateRequest(self._service.url, self._service.connection, create_entity_handler, self._entity_set,
self.last_segment)

def update_entity(self, key=None, **kwargs):
def update_entity(self, key=None, **kwargs) -> EntityModifyRequest:
"""Updates an existing entity in the given entity-set."""

def update_entity_handler(response):
def update_entity_handler(response: requests.Response) -> None:
"""Gets modified entity encoded in HTTP Response"""

if response.status_code != 204:
Expand Down Expand Up @@ -1343,7 +1347,7 @@ def http_get(self, path: str, connection: Optional[requests.Session] = None) ->
return conn.get(urljoin(self._url, path))

def http_get_odata(self, path: str, handler: Callable[[requests.Response], Any],
connection: Optional[requests.Session] = None):
connection: Optional[requests.Session] = None) -> ODataHttpRequest:
"""HTTP GET request proxy for the passed path in the service"""

conn = connection
Expand All @@ -1364,7 +1368,7 @@ def batch_handler(batch: MultipartRequest, parts: List[List[str]]) -> List[Any]:

logging.getLogger(LOGGER_NAME).debug('Batch handler called for batch %s', batch.id)

result = []
result: List[Any] = []
for part, req in zip(parts, batch.requests):
logging.getLogger(LOGGER_NAME).debug('Batch handler is processing part %s for request %s', part, req)

Expand All @@ -1384,7 +1388,7 @@ def batch_handler(batch: MultipartRequest, parts: List[List[str]]) -> List[Any]:
def create_changeset(self, changeset_id=None):
"""Create instance of OData changeset"""

def changeset_handler(changeset: 'Changeset', parts: List[str]):
def changeset_handler(changeset: 'Changeset', parts: List[str]) -> List[ODataHttpResponse]:
"""Gets changeset response from HTTP response"""

logging.getLogger(LOGGER_NAME).debug('Changeset handler called for changeset %s', changeset.id)
Expand Down

0 comments on commit f160c93

Please sign in to comment.