8
8
import hmac
9
9
import base64
10
10
11
- try :
12
- from urllib .parse import quote
13
- except ImportError :
14
- from urllib2 import quote # type: ignore
11
+ from urllib .parse import quote
15
12
16
13
from azure .core .pipeline .transport import HttpRequest
17
14
from azure .core .pipeline .policies import AzureKeyCredentialPolicy , BearerTokenCredentialPolicy
27
24
if TYPE_CHECKING :
28
25
from datetime import datetime
29
26
30
- def generate_sas (endpoint , shared_access_key , expiration_date_utc , ** kwargs ):
31
- # type: (str, str, datetime, Any) -> str
27
+ def generate_sas (
28
+ endpoint : str ,
29
+ shared_access_key : str ,
30
+ expiration_date_utc : "datetime" ,
31
+ * ,
32
+ api_version : str = constants .DEFAULT_API_VERSION ,
33
+ ) -> str :
32
34
"""Helper method to generate shared access signature given hostname, key, and expiration date.
33
35
:param str endpoint: The topic endpoint to send the events to.
34
36
Similar to <YOUR-TOPIC-NAME>.<YOUR-REGION-NAME>-1.eventgrid.azure.net
@@ -50,7 +52,7 @@ def generate_sas(endpoint, shared_access_key, expiration_date_utc, **kwargs):
50
52
:caption: Generate a shared access signature.
51
53
"""
52
54
full_endpoint = "{}?apiVersion={}" .format (
53
- endpoint , kwargs . get ( " api_version" , constants . DEFAULT_API_VERSION )
55
+ endpoint , api_version
54
56
)
55
57
encoded_resource = quote (full_endpoint , safe = constants .SAFE_ENCODE )
56
58
encoded_expiration_utc = quote (str (expiration_date_utc ), safe = constants .SAFE_ENCODE )
0 commit comments