File: preparers_async.py

package info (click to toggle)
python-azure 20250603%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 851,724 kB
  • sloc: python: 7,362,925; ansic: 804; javascript: 287; makefile: 195; sh: 145; xml: 109
file content (57 lines) | stat: -rw-r--r-- 1,979 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
# coding: utf-8

# -------------------------------------------------------------------------
# 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 Awaitable, List
from azure.security.attestation import AttestationType

try:
    from typing import TYPE_CHECKING
except ImportError:
    TYPE_CHECKING = False

from typing import Awaitable, Callable, Dict, Optional, Any, TypeVar, overload

T = TypeVar("T")


def AllAttestationTypes(func: Callable[..., Awaitable[T]] = None, **kwargs: Any):
    """Decorator to apply to function to add attestation_type kwarg for each attestation type."""

    async def wrapper(*args, **kwargs) -> Callable[..., Awaitable[T]]:
        for attestation_type in [
            AttestationType.SGX_ENCLAVE,
            AttestationType.OPEN_ENCLAVE,
            AttestationType.TPM,
        ]:
            await func(*args, attestation_type=attestation_type, **kwargs)

    return wrapper


def AllInstanceTypes(
    func: Callable[..., Awaitable[T]] = None, include_shared: bool = True, **kwargs: Any
):
    """Decorator to apply to function to add instance_url kwarg for each instance type."""

    async def wrapper(*args, **kwargs) -> Callable[..., Awaitable[T]]:
        instances = []  # type:List[str]
        instances.append(kwargs.get("attestation_aad_url"))
        instances.append(kwargs.get("attestation_isolated_url"))
        if include_shared:
            instances.append(
                "https://shared"
                + kwargs.get("attestation_location_short_name")
                + "."
                + kwargs.get("attestation_location_short_name")
                + ".attest.azure.net"
            )

        for instance_url in instances:
            await func(*args, instance_url=instance_url, **kwargs)

    return wrapper