From 1ea302aac62f9ddf9af589ef1e65858ba7800c0d Mon Sep 17 00:00:00 2001 From: Johann Kellerman Date: Wed, 4 Oct 2023 09:46:04 +0200 Subject: [PATCH] typing_Entitycollection --- office365/entity_collection.py | 12 ++-- office365/graph_client.py | 19 +++--- office365/onedrive/drives/drive.py | 1 + office365/onedrive/sites/site.py | 24 +++++++- office365/onedrive/sites/sites_with_root.py | 2 + office365/runtime/client_object_collection.py | 61 ++++++++----------- 6 files changed, 69 insertions(+), 50 deletions(-) diff --git a/office365/entity_collection.py b/office365/entity_collection.py index 55c3aa71..17e3d968 100644 --- a/office365/entity_collection.py +++ b/office365/entity_collection.py @@ -1,14 +1,19 @@ +from typing import TypeVar + from office365.runtime.client_object_collection import ClientObjectCollection from office365.runtime.compat import is_string_type from office365.runtime.paths.item import ItemPath from office365.runtime.queries.create_entity import CreateEntityQuery from office365.runtime.paths.resource_path import ResourcePath +from office365.graph_client import GraphClient +T = TypeVar("T") -class EntityCollection(ClientObjectCollection): +class EntityCollection(ClientObjectCollection[T]): """A collection container which represents a named collections of entities""" def __getitem__(self, key): + # type: (int | str) -> T """ :param key: key is used to address an entity by either an index or by identifier :type key: int or str @@ -21,6 +26,7 @@ def __getitem__(self, key): raise ValueError("Invalid key: expected either an entity index [int] or identifier [str]") def add(self, **kwargs): + # type: (Any) -> T """ Creates an entity and prepares the query """ @@ -32,7 +38,5 @@ def add(self, **kwargs): @property def context(self): - """ - :rtype: office365.graph_client.GraphClient - """ + # type: () -> GraphClient return self._context diff --git a/office365/graph_client.py b/office365/graph_client.py index 1ba0d615..a31541b7 100644 --- a/office365/graph_client.py +++ b/office365/graph_client.py @@ -1,3 +1,5 @@ +from typing import Callable + from office365.booking.solutions.root import SolutionsRoot from office365.communications.cloud_communications import CloudCommunications from office365.delta_collection import DeltaCollection @@ -64,6 +66,7 @@ class GraphClient(ClientRuntimeContext): """Graph Service client""" def __init__(self, acquire_token_callback): + # type: (Callable[None, dict]) -> None """ :param () -> dict acquire_token_callback: Acquire token function """ @@ -86,6 +89,7 @@ def execute_batch(self, items_per_batch=100): return self def pending_request(self): + # type: () -> ODataRequest if self._pending_request is None: self._pending_request = ODataRequest(V4JsonFormat()) self._pending_request.beforeExecute += self._authenticate_request @@ -93,25 +97,20 @@ def pending_request(self): return self._pending_request def service_root_url(self): + # type: () -> str return "https://graph.microsoft.com/v1.0" def _build_specific_query(self, request): - """ - Builds Graph specific HTTP request - - :type request: RequestOptions - """ + # type: (RequestOptions) -> None + """Builds Graph specific HTTP request.""" if isinstance(self.current_query, UpdateEntityQuery): request.method = HttpMethod.Patch elif isinstance(self.current_query, DeleteEntityQuery): request.method = HttpMethod.Delete def _authenticate_request(self, request): - """ - Authenticate request - - :type request: RequestOptions - """ + # type: (RequestOptions) -> None + """Authenticate request.""" token_json = self._acquire_token_callback() token = TokenResponse.from_json(token_json) request.ensure_header('Authorization', 'Bearer {0}'.format(token.accessToken)) diff --git a/office365/onedrive/drives/drive.py b/office365/onedrive/drives/drive.py index 3a522def..a871a98e 100644 --- a/office365/onedrive/drives/drive.py +++ b/office365/onedrive/drives/drive.py @@ -95,6 +95,7 @@ def owner(self): @property def root(self): + # type: () -> DriveItem """The root folder of the drive.""" return self.properties.get('root', DriveItem(self.context, RootPath(self.resource_path, self.items.resource_path))) diff --git a/office365/onedrive/sites/site.py b/office365/onedrive/sites/site.py index 5ab62845..00f071df 100644 --- a/office365/onedrive/sites/site.py +++ b/office365/onedrive/sites/site.py @@ -1,3 +1,6 @@ +from typing import Optional +from datetime import datetime + from office365.base_item import BaseItem from office365.entity_collection import EntityCollection from office365.onedrive.analytics.item_activity_stat import ItemActivityStat @@ -22,6 +25,7 @@ class Site(BaseItem): """The site resource provides metadata and relationships for a SharePoint site. """ def get_by_path(self, path): + # type: (str) -> Site """ Retrieve properties and relationships for a site resource. A site resource represents a team site in SharePoint. @@ -34,8 +38,6 @@ def get_by_path(self, path): /sites/root: The tenant root site. /groups/{group-id}/sites/root: The group's team site. - - :type path: str """ return_type = Site(self.context) qry = ServiceOperationQuery(self, "GetByPath", [path], None, None, return_type) @@ -43,6 +45,7 @@ def get_by_path(self, path): return return_type def get_applicable_content_types_for_list(self, list_id): + # type: (str) -> ContentTypeCollection """ Get site contentTypes that can be added to a list. @@ -57,6 +60,7 @@ def get_applicable_content_types_for_list(self, list_id): return return_type def get_activities_by_interval(self, start_dt=None, end_dt=None, interval=None): + # type: (Optional[datetime], Optional[datetime], str) -> EntityCollection[ItemActivityStat] """ Get a collection of itemActivityStats resources for the activities that took place on this resource within the specified time interval. @@ -77,22 +81,26 @@ def get_activities_by_interval(self, start_dt=None, end_dt=None, interval=None): @property def site_collection(self): + # type: () -> SiteCollection """Provides details about the site's site collection. Available only on the root site.""" return self.properties.get("siteCollection", SiteCollection()) @property def sharepoint_ids(self): + # type: () -> SharePointIds """Returns identifiers useful for SharePoint REST compatibility.""" return self.properties.get('sharepointIds', SharePointIds()) @property def items(self): + # type: () -> EntityCollection[ListItem] """Used to address any item contained in this site. This collection cannot be enumerated.""" return self.properties.get('items', EntityCollection(self.context, ListItem, ResourcePath("items", self.resource_path))) @property def columns(self): + # type: () -> ColumnDefinitionCollection """The collection of columns under this site.""" return self.properties.get('columns', ColumnDefinitionCollection(self.context, @@ -100,6 +108,7 @@ def columns(self): @property def external_columns(self): + # type: () -> ColumnDefinitionCollection """The collection of columns under this site.""" return self.properties.get('externalColumns', ColumnDefinitionCollection(self.context, @@ -108,6 +117,7 @@ def external_columns(self): @property def content_types(self): + # type: () -> ContentTypeCollection """The collection of content types under this site.""" return self.properties.get('contentTypes', ContentTypeCollection(self.context, @@ -115,12 +125,14 @@ def content_types(self): @property def lists(self): + # type: () -> ListCollection """The collection of lists under this site.""" return self.properties.get('lists', ListCollection(self.context, ResourcePath("lists", self.resource_path))) @property def operations(self): + # type: () -> EntityCollection[RichLongRunningOperation] """The collection of long-running operations on the site.""" return self.properties.get('operations', EntityCollection(self.context, RichLongRunningOperation, @@ -128,48 +140,56 @@ def operations(self): @property def permissions(self): + # type: () -> PermissionCollection """The permissions associated with the site.""" return self.properties.get('permissions', PermissionCollection(self.context, ResourcePath("permissions", self.resource_path))) @property def drive(self): + # type: () -> Drive """The default drive (document library) for this site.""" return self.properties.get('drive', Drive(self.context, ResourcePath("drive", self.resource_path))) @property def drives(self): + # type: () -> EntityCollection[Drive] """The collection of drives under this site.""" return self.properties.get('drives', EntityCollection(self.context, Drive, ResourcePath("drives", self.resource_path))) @property def sites(self): + # type: () -> EntityCollection[Site] """The collection of sites under this site.""" return self.properties.get('sites', EntityCollection(self.context, Site, ResourcePath("sites", self.resource_path))) @property def analytics(self): + # type: () -> ItemAnalytics """Analytics about the view activities that took place on this site.""" return self.properties.get('analytics', ItemAnalytics(self.context, ResourcePath("analytics", self.resource_path))) @property def onenote(self): + # type: () -> Onenote """Represents the Onenote services available to a site.""" return self.properties.get('onenote', Onenote(self.context, ResourcePath("onenote", self.resource_path))) @property def term_store(self): + # type: () -> Store """The default termStore under this site.""" return self.properties.get('termStore', Store(self.context, ResourcePath("termStore", self.resource_path))) @property def term_stores(self): + # type: () -> EntityCollection[Store] """The collection of termStores under this site.""" return self.properties.get('termStores', EntityCollection(self.context, Store, diff --git a/office365/onedrive/sites/sites_with_root.py b/office365/onedrive/sites/sites_with_root.py index 8d7782dd..95bdbb6b 100644 --- a/office365/onedrive/sites/sites_with_root.py +++ b/office365/onedrive/sites/sites_with_root.py @@ -46,6 +46,7 @@ def remove(self, sites): return return_type def search(self, query_text): + # type: (str) -> SitesWithRoot """ Search across a SharePoint tenant for sites that match keywords provided. @@ -61,4 +62,5 @@ def search(self, query_text): @property def root(self): + # type: () -> Site return self.properties.get('root', Site(self.context, RootPath(self.resource_path, self.resource_path))) diff --git a/office365/runtime/client_object_collection.py b/office365/runtime/client_object_collection.py index a71a610f..beeabf68 100644 --- a/office365/runtime/client_object_collection.py +++ b/office365/runtime/client_object_collection.py @@ -1,24 +1,21 @@ -from typing import TypeVar, Iterator +from typing import TypeVar, Optional, Generic, Iterator +from typing_extensions import Self from office365.runtime.client_object import ClientObject from office365.runtime.types.event_handler import EventHandler +from office365.runtime.client_runtime_context import ClientRuntimeContext +from office365.runtime.paths.resource_path import ResourcePath -T = TypeVar('T', bound='ClientObjectCollection') -P_T = TypeVar("P_T", bound=ClientObject) +T = TypeVar("T") -class ClientObjectCollection(ClientObject): +class ClientObjectCollection(ClientObject, Generic[T]): def __init__(self, context, item_type, resource_path=None, parent=None): - """A collection container which represents a named collections of objects - - :type context: office365.runtime.client_runtime_context.ClientRuntimeContext - :type item_type: type[ClientObject] - :type resource_path: office365.runtime.paths.resource_path.ResourcePath - :type parent: ClientObject or None - """ + # type: (ClientRuntimeContext, T, Optional[ResourcePath], Optional[ClientObject]) -> None + """A collection container which represents a named collections of objects.""" super(ClientObjectCollection, self).__init__(context, resource_path) - self._data = [] # type: list[ClientObject] + self._data = [] # type: list[T] self._item_type = item_type self._page_loaded = EventHandler(False) self._paged_mode = False @@ -35,11 +32,8 @@ def clear(self): return self def create_typed_object(self, initial_properties=None, resource_path=None): - """ - :type self: T - :type initial_properties: dict or None - :type resource_path: office365.runtime.paths.resource_path.ResourcePath or None - """ + # type: (Optional[dict], Optional[ResourcePath]) -> T + """Create an object from the item_type.""" if self._item_type is None: raise AttributeError("No class model for entity type '{0}' was found".format(self._item_type)) client_object = self._item_type(context=self.context, resource_path=resource_path) # type: ClientObject @@ -48,11 +42,7 @@ def create_typed_object(self, initial_properties=None, resource_path=None): return client_object def set_property(self, key, value, persist_changes=False): - """ - :type key: int or str - :type value: dict - :type persist_changes: bool - """ + # type: (str | int, dict, bool) -> Self if key == "__nextLinkUrl": self._next_request_url = value else: @@ -62,6 +52,7 @@ def set_property(self, key, value, persist_changes=False): return self def add_child(self, client_object): + # type: (T) -> Self """ Adds client object into collection @@ -72,17 +63,12 @@ def add_child(self, client_object): return self def remove_child(self, client_object): - """ - :type client_object: ClientObject - """ + # type: (T) -> Self self._data = [item for item in self._data if item != client_object] return self def __iter__(self): - """ - :type self: T - :rtype: Iterator[ClientObject] - """ + # type: () -> Iterator[T] for item in self._data: yield item if self._paged_mode: @@ -93,26 +79,27 @@ def __iter__(self): yield next_item def __len__(self): + # type: () -> int return len(self._data) def __repr__(self): + # type: () -> str return repr(self._data) def __getitem__(self, index): - """ - :type index: int - """ + # type: (int) -> T return self._data[index] def to_json(self, json_format=None): - """Serializes the collection into JSON""" + # type: (int) -> List[dict] + """Serializes the collection into JSON.""" return [item.to_json(json_format) for item in self._data] def filter(self, expression): + # type: (str) -> Self """ Allows clients to filter a collection of resources that are addressed by a request URL - :type self: T :param str expression: Filter expression, for example: 'Id eq 123' """ self.query_options.filter = expression @@ -191,6 +178,7 @@ def _page_loaded(items): return self def _get_next(self, after_loaded=None): + # type: (Optional[EventHandler]) -> Self """ Submit a request to retrieve next collection of items @@ -207,6 +195,7 @@ def _construct_next_query(request): return self def first(self, expression): + # type: (str) -> T """Return the first Entity instance that matches current query :param str expression: Filter expression @@ -229,6 +218,7 @@ def _after_loaded(col): return return_type def single(self, expression): + # type: (str) -> T """ Return only one resulting Entity @@ -256,15 +246,18 @@ def _after_loaded(col): @property def parent(self): + # type: () -> ClientObject return self._parent @property def has_next(self): + # type: () -> bool """""" return self._next_request_url is not None @property def entity_type_name(self): + # type: () -> str """Returns server type name for the collection of entities""" if self._entity_type_name is None: client_object = self.create_typed_object()