[EG] next-mypy (Azure#31857)

* next-mypy

* fix analyze

* update typing

* update comments

* remove Any

Author: l0lawrence

sdk/eventgrid/azure-eventgrid/README.md

@@ -60,9 +60,9 @@ For example, you can use `DefaultAzureCredential` to construct a client which wi
 from azure.identity import DefaultAzureCredential
 from azure.eventgrid import EventGridPublisherClient, EventGridEvent
 
-credential = DefaultAzureCredential()
+default_az_credential = DefaultAzureCredential()
 endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
-client = EventGridPublisherClient(endpoint, credential)
+client = EventGridPublisherClient(endpoint, default_az_credential)
 ```
 
 <!-- END SNIPPET -->
@@ -88,8 +88,8 @@ from azure.core.credentials import AzureKeyCredential
 topic_key = os.environ["EVENTGRID_TOPIC_KEY"]
 endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
 
-credential = AzureKeyCredential(topic_key)
-client = EventGridPublisherClient(endpoint, credential)
+credential_key = AzureKeyCredential(topic_key)
+client = EventGridPublisherClient(endpoint, credential_key)
 ```
 
 <!-- END SNIPPET -->

sdk/eventgrid/azure-eventgrid/azure/eventgrid/_helpers.py

@@ -2,7 +2,7 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Dict
 import json
 import hashlib
 import hmac
@@ -165,13 +165,13 @@ def _from_cncf_events(event): # pylint: disable=inconsistent-return-statements
 
 def _build_request(endpoint, content_type, events, *, channel_name=None):
     serialize = Serializer()
-    header_parameters = {}  # type: Dict[str, Any]
+    header_parameters: Dict[str, Any] = {}
     header_parameters['Content-Type'] = serialize.header("content_type", content_type, 'str')
 
     if channel_name:
         header_parameters['aeg-channel-name'] = channel_name
 
-    query_parameters = {}  # type: Dict[str, Any]
+    query_parameters: Dict[str, Any] = {}
     query_parameters['api-version'] = serialize.query("api_version", "2018-01-01", 'str')
 
     body = serialize.body(events, '[object]')

sdk/eventgrid/azure-eventgrid/azure/eventgrid/_publisher_client.py

@@ -54,13 +54,17 @@
         TokenCredential,
     )
 
+    from cloudevents.http.event import CloudEvent as CNCFCloudEvent
+
 SendType = Union[
     CloudEvent,
     EventGridEvent,
     Dict,
+    "CNCFCloudEvent",
     List[CloudEvent],
     List[EventGridEvent],
     List[Dict],
+    List["CNCFCloudEvent"],
 ]
 
 ListEventType = Union[List[CloudEvent], List[EventGridEvent], List[Dict]]

sdk/eventgrid/azure-eventgrid/azure/eventgrid/aio/_publisher_client_async.py

@@ -48,8 +48,17 @@
 if TYPE_CHECKING:
     from azure.core.credentials_async import AsyncTokenCredential
 
+    from cloudevents.http.event import CloudEvent as CNCFCloudEvent
+
 SendType = Union[
-    CloudEvent, EventGridEvent, Dict, List[CloudEvent], List[EventGridEvent], List[Dict]
+    CloudEvent,
+    EventGridEvent,
+    Dict,
+    "CNCFCloudEvent",
+    List[CloudEvent],
+    List[EventGridEvent],
+    List[Dict],
+    List["CNCFCloudEvent"],
 ]
 
 ListEventType = Union[List[CloudEvent], List[EventGridEvent], List[Dict]]

sdk/eventgrid/azure-eventgrid/samples/async_samples/sample_authentication_async.py

@@ -23,8 +23,8 @@
 topic_key = os.environ["EVENTGRID_TOPIC_KEY"]
 endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
 
-credential = AzureKeyCredential(topic_key)
-client = EventGridPublisherClient(endpoint, credential)
+credential_key = AzureKeyCredential(topic_key)
+client = EventGridPublisherClient(endpoint, credential_key)
 # [END client_auth_with_key_cred_async]
 
 # [START client_auth_with_sas_cred_async]
@@ -35,8 +35,8 @@
 signature = os.environ["EVENTGRID_SAS"]
 endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
 
-credential = AzureSasCredential(signature)
-client = EventGridPublisherClient(endpoint, credential)
+credential_sas = AzureSasCredential(signature)
+client = EventGridPublisherClient(endpoint, credential_sas)
 # [END client_auth_with_sas_cred_async]
 
 # [START client_auth_with_token_cred_async]
@@ -51,7 +51,7 @@
     data_version="2.0"
 )
 
-credential = DefaultAzureCredential()
+default_az_credential = DefaultAzureCredential()
 endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
-client = EventGridPublisherClient(endpoint, credential)
+client = EventGridPublisherClient(endpoint, default_az_credential)
 # [END client_auth_with_token_cred_async]

sdk/eventgrid/azure-eventgrid/samples/consume_samples/consume_cloud_events_from_eventhub.py

@@ -25,7 +25,7 @@
 CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"]
 EVENTHUB_NAME = os.environ["EVENT_HUB_NAME"]
 def on_event(partition_context, event):
-    dict_event = CloudEvent.from_json(event)
+    dict_event: CloudEvent = CloudEvent.from_json(event)
     print("data: {}\n".format(dict_event.data))
 
 consumer_client = EventHubConsumerClient.from_connection_string(

sdk/eventgrid/azure-eventgrid/samples/consume_samples/consume_cloud_events_from_storage_queue.py

@@ -13,7 +13,7 @@
     1) STORAGE_QUEUE_CONN_STR: The connection string to the Storage account
     3) STORAGE_QUEUE_NAME: The name of the storage queue.
 """
-
+from typing import List
 from azure.core.messaging import CloudEvent
 from azure.storage.queue import QueueServiceClient, BinaryBase64DecodePolicy
 import os
@@ -30,7 +30,7 @@
         ).peek_messages(max_messages=32)
 
     ## deserialize payload into a list of typed Events
-    events = [CloudEvent.from_json(msg) for msg in payload]
+    events: List[CloudEvent] = [CloudEvent.from_json(msg) for msg in payload]
 
     for event in events:
         print(type(event)) ## CloudEvent

sdk/eventgrid/azure-eventgrid/samples/publish_samples/publish_cloud_events_to_custom_topic_sample.py

@@ -23,7 +23,7 @@
 from azure.core.messaging import CloudEvent
 from azure.eventgrid import EventGridPublisherClient
 
-key = os.environ.get("EVENTGRID_CLOUD_EVENT_TOPIC_KEY")
+key = os.environ["EVENTGRID_CLOUD_EVENT_TOPIC_KEY"]
 endpoint = os.environ["EVENTGRID_CLOUD_EVENT_TOPIC_ENDPOINT"]
 
 # authenticate client

sdk/eventgrid/azure-eventgrid/samples/sync_samples/sample_authentication.py

@@ -23,8 +23,8 @@
 topic_key = os.environ["EVENTGRID_TOPIC_KEY"]
 endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
 
-credential = AzureKeyCredential(topic_key)
-client = EventGridPublisherClient(endpoint, credential)
+credential_key = AzureKeyCredential(topic_key)
+client = EventGridPublisherClient(endpoint, credential_key)
 # [END client_auth_with_key_cred]
 
 # [START client_auth_with_sas_cred]
@@ -35,15 +35,15 @@
 signature = os.environ["EVENTGRID_SAS"]
 endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
 
-credential = AzureSasCredential(signature)
-client = EventGridPublisherClient(endpoint, credential)
+credential_sas = AzureSasCredential(signature)
+client = EventGridPublisherClient(endpoint, credential_sas)
 # [END client_auth_with_sas_cred]
 
 # [START client_auth_with_token_cred]
 from azure.identity import DefaultAzureCredential
 from azure.eventgrid import EventGridPublisherClient, EventGridEvent
 
-credential = DefaultAzureCredential()
+default_az_credential = DefaultAzureCredential()
 endpoint = os.environ["EVENTGRID_TOPIC_ENDPOINT"]
-client = EventGridPublisherClient(endpoint, credential)
+client = EventGridPublisherClient(endpoint, default_az_credential)
 # [END client_auth_with_token_cred]

sdk/eventgrid/azure-eventgrid/samples/sync_samples/sample_consume_custom_payload.py

@@ -6,12 +6,12 @@
 """
 FILE: sample_consume_custom_payload.py
 DESCRIPTION:
-    These samples demonstrate consuming a raw json with a list of jsons and
+    These samples demonstrate consuming a raw json with a list of json and
     deserializing them into typed objects of CloudEvents.
 USAGE:
     python sample_consume_custom_payload.py
 """
-
+from typing import List
 from azure.core.messaging import CloudEvent
 import json
 
@@ -27,7 +27,7 @@
     "specversion":"1.0"
 }]"""
 
-deserialized_dict_events = [CloudEvent(**msg) for msg in json.loads(cloud_custom_dict)]
+deserialized_dict_events: List[CloudEvent] = [CloudEvent(**msg) for msg in json.loads(cloud_custom_dict)]
 
 for event in deserialized_dict_events:
     print(event.data)