partition_service.py 2.13 KB
Newer Older
Luc Yriarte's avatar
Luc Yriarte committed
1
2
3
4
5
6
from os import environ

from osdu_az.opendes.fake_partition_client import FakePartitionClient
from osdu_az.opendes.partition_client import PartitionClient
from osdu_az.partition.partition_info import PartitionInfo
from osdu_az.partition.partitions_cache import PartitionsCache
Yannick's avatar
Yannick committed
7
8
9
import logging

_LOGGER = logging.getLogger('osdu_az.storage.blob_storage_az')
Luc Yriarte's avatar
Luc Yriarte committed
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# KNOWN CORE PARTITION PROPERTIES
COSMOS_ENDPOINT = 'cosmos-endpoint'
COSMOS_CONNECTION = 'cosmos-connection'
ELASTIC_USERNAME = 'elastic-username'
ELASTIC_ENDPOINT = 'elastic-endpoint'
SB_CONNECTION = 'sb-connection'
COSMOS_PRIMARY_KEY = 'cosmos-primary-key'
COMPLIANCE_RULESET = 'compliance-ruleset'
STORAGE_ACCOUNT_NAME = 'storage-account-name'
ELASTIC_PASSWORD = 'elastic-password'
STORAGE_ACCOUNT_KEY = 'storage-account-key'
SB_NAMESPACE = 'sb-namespace'
PARTITION_ID = 'id'


class PartitionService:
    partition_client = None
    scope = ''

    @staticmethod
    async def get_partition(data_partition_id: str) -> PartitionInfo:
        partition_info = PartitionsCache.get(data_partition_id)
        if not partition_info:
Yannick's avatar
Yannick committed
34
            _LOGGER.info(f'PartitionService fetching partition info for "{data_partition_id}"')
Luc Yriarte's avatar
Luc Yriarte committed
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
            partition_info = await PartitionService._partition_client().get_partition(data_partition_id)
            if partition_info:
                PartitionsCache.set(data_partition_id, partition_info)
        return partition_info

    @staticmethod
    async def get_storage_account_name(data_partition_id: str):
        partition_info = await PartitionService.get_partition(data_partition_id)
        if partition_info:
            return partition_info.get_value(STORAGE_ACCOUNT_NAME)

    @staticmethod
    def _partition_client():
        if not PartitionService.partition_client:
            PartitionService._initialize_partition_client()
        return PartitionService.partition_client

    @staticmethod
    def _initialize_partition_client():
        if environ.get('USE_PARTITION_SERVICE', default='enabled').lower() == 'enabled':
            PartitionService.partition_client = PartitionClient()
        else:
            PartitionService.partition_client = FakePartitionClient()