File: async_proxy.py

package info (click to toggle)
python-azure 20201208%2Bgit-6
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 1,437,920 kB
  • sloc: python: 4,287,452; javascript: 269; makefile: 198; sh: 187; xml: 106
file content (82 lines) | stat: -rw-r--r-- 2,654 bytes parent folder | download
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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
from asyncio import get_event_loop
from azure.appconfiguration.aio import AzureAppConfigurationClient


def _to_list(ait):
    async def lst():
        result = []
        async for item in ait:
            result.append(item)
        return result

    return get_event_loop().run_until_complete(lst())


class AzureAppConfigurationClientProxy(object):
    @classmethod
    def from_connection_string(
        cls,
        connection_string,  # type: str
        **kwargs
    ):
        return cls(
            AzureAppConfigurationClient.from_connection_string(
                connection_string, **kwargs
            )
        )

    def __init__(self, obj):
        self.obj = obj

    def get_configuration_setting(
        self, key, label=None, accept_datetime=None, **kwargs
    ):
        return get_event_loop().run_until_complete(
            self.obj.get_configuration_setting(
                key, label=label, accept_datetime=accept_datetime, **kwargs
            )
        )

    def add_configuration_setting(self, configuration_setting, **kwargs):
        return get_event_loop().run_until_complete(
            self.obj.add_configuration_setting(configuration_setting, **kwargs)
        )

    def delete_configuration_setting(self, key, label=None, etag=None, **kwargs):
        return get_event_loop().run_until_complete(
            self.obj.delete_configuration_setting(key, label=label, etag=etag, **kwargs)
        )

    def list_configuration_settings(
        self, label_filter=None, key_filter=None, accept_datetime=None, fields=None, **kwargs
    ):
        paged = self.obj.list_configuration_settings(
            label_filter=label_filter,
            key_filter=key_filter,
            accept_datetime=accept_datetime,
            fields=fields,
            **kwargs
        )
        return _to_list(paged)

    def list_revisions(
        self, label_filter=None, key_filter=None, accept_datetime=None, fields=None, **kwargs
    ):
        paged = self.obj.list_revisions(
            label_filter=label_filter,
            key_filter=key_filter,
            accept_datetime=accept_datetime,
            fields=fields,
            **kwargs
        )
        return _to_list(paged)

    def set_configuration_setting(self, configuration_setting, **kwargs):
        return get_event_loop().run_until_complete(
            self.obj.set_configuration_setting(configuration_setting, **kwargs)
        )

    def set_read_only(self, configuration_setting, read_only=True, **kwargs):
        return get_event_loop().run_until_complete(
            self.obj.set_read_only(configuration_setting, read_only, **kwargs)
        )