Inventory 플러그인 개발

from spaceone.inventory.plugin.collector.lib.server import CollectorPluginServer

app = CollectorPluginServer()


@app.route("Collector.init")
def collector_init(params: dict) -> dict:
    """init plugin by options

    Args:
        params (CollectorInitRequest): {
            'options': 'dict',    # Required
            'domain_id': 'str'
        }

    Returns:
        PluginResponse: {
            'metadata': 'dict'
        }
    """
    return {"metadata": {"options_schema": {}}}

import logging

from spaceone.core.manager import BaseManager

_LOGGER = logging.getLogger(__name__)


class CryptoManager(BaseManager):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def collect_resources(self, options, secret_data, schema):
        pass

from typing import Generator

from spaceone.inventory.plugin.collector.lib.server import CollectorPluginServer

from plugin.manager.crypto_manager import CryptoManager

...

@app.route("Collector.collect")
def collector_collect(params: dict) -> Generator[dict, None, None]:
    ...
    
    options = params["options"]
    secret_data = params["secret_data"]
    schema = params.get("schema")

    crypto_mgr = CryptoManager()
    return crypto_mgr.collect_resources(options, secret_data, schema)

import logging
import os

from spaceone.core.manager import BaseManager

_LOGGER = logging.getLogger(__name__)
_CURRENT_DIR = os.path.dirname(__file__)
_METADATA_DIR = os.path.join(_CURRENT_DIR, "../metadata/")


class CryptoManager(BaseManager):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.provider = "portfolio"
        self.cloud_service_group = "Investment"
        self.cloud_service_type = "Cryptocurrency"
        self.metadata_path = os.path.join(
            _METADATA_DIR, "investment/cryptocurrency.yaml"
        )
    ...

import logging

from spaceone.core.manager import BaseManager
from spaceone.inventory.plugin.collector.lib import (
    make_cloud_service_type,
    make_error_response,
    make_response,
)

_LOGGER = logging.getLogger(__name__)


class CryptoManager(BaseManager):
    ...
     
    def collect_resources(self, options, secret_data, schema):
        try:
            yield from self.collect_cloud_service_type(options, secret_data, schema)
            yield from self.collect_cloud_service(options, secret_data, schema)
        except Exception as e:
            yield make_error_response(
                error=e,
                provider=self.provider,
                cloud_service_group=self.cloud_service_group,
                cloud_service_type=self.cloud_service_type,
            )

    def collect_cloud_service_type(self, options, secret_data, schema):
        cloud_service_type = make_cloud_service_type(
            name=self.cloud_service_type,
            group=self.cloud_service_group,
            provider=self.provider,
            metadata_path=self.metadata_path,
            is_primary=True,
            is_major=True,
        )

        yield make_response(
            cloud_service_type=cloud_service_type,
            match_keys=[["name", "reference.resource_id", "account", "provider"]],
            resource_type="inventory.CloudServiceType",
        )

    def collect_cloud_service(self, options, secret_data, schema):
        pass

import logging
from typing import Dict, List

from pycoingecko import CoinGeckoAPI
from spaceone.core.connector import BaseConnector

_LOGGER = logging.getLogger(__name__)


class CryptoConnector(BaseConnector):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.client = CoinGeckoAPI()

    def list_cryptocurrencies(self) -> List[Dict]:
        try:
            coins = self.client.get_coins_markets(
                vs_currency="krw", order="market_cap_desc", per_page=5, page=1
            )

            filtered_coins = list()
            for coin in coins:
                filtered_coins.append(
                    {
                        "name": coin["name"],
                        "current_price": coin["current_price"],
                        "market_cap_rank": coin["market_cap_rank"],
                        "price_change_percentage_24h": coin[
                            "price_change_percentage_24h"
                        ],
                        "high_24h": coin["high_24h"],
                        "low_24h": coin["low_24h"],
                        "last_updated": coin["last_updated"],
                    }
                )
            return filtered_coins
        except Exception as e:
            _LOGGER.error(f"Error fetching cryptocurrency data: {e}")
            return []

...

from spaceone.inventory.plugin.collector.lib import (
    make_cloud_service_type,
    make_cloud_service_with_metadata,
    make_error_response,
    make_response,
)

from plugin.connector.cryptocurrency_connector import CryptoConnector

...

class CryptoManager(BaseManager):
    ...
    
    def collect_cloud_service(self, options, secret_data, schema):
        crypto_connector = CryptoConnector()
        cryptocurrencies = crypto_connector.list_cryptocurrencies()

        for crypto in cryptocurrencies:
            cloud_service = make_cloud_service_with_metadata(
                name=crypto["name"],
                cloud_service_type=self.cloud_service_type,
                cloud_service_group=self.cloud_service_group,
                provider=self.provider,
                data=crypto,
                data_format="dict",
                metadata_path=self.metadata_path,
            )
            yield make_response(
                cloud_service=cloud_service,
                match_keys=[["name", "reference.resource_id", "account", "provider"]],
            )

search:
  fields:
    - 이름: data.name
    - 시가총액 순위: data.market_cap_rank
table:
  sort:
    key: data.market_cap_rank
    desc: true
  fields:
    - 이름: data.name
    - 가격: data.current_price
    - 가격 업데이트 시간 (UTC): data.last_updated
    - 시가총액 순위: data.market_cap_rank
    - 24시간 변동률 (%): data.price_change_percentage_24h
    - 24시간 최저가: data.low_24h
    - 24시간 최고가: data.high_24h
tabs.0:
  name: Details
  type: item
  fields:
    - 이름: data.name
    - 가격: data.current_price
    - 가격 업데이트 시간 (UTC): data.last_updated
    - 시가총액 순위: data.market_cap_rank
    - 24시간 변동률 (%): data.price_change_percentage_24h
    - 24시간 최저가: data.low_24h
    - 24시간 최고가: data.high_24h

2024-07-16T10:48:44.515Z [DEBUG]       (server.py:69) Loaded Services: 
         - spaceone.api.inventory.plugin.Collector
         - spaceone.api.inventory.plugin.Job
         - grpc.health.v1.Health
         - spaceone.api.core.v1.ServerInfo  
2024-07-16T10:48:44.515Z [INFO]       (server.py:73) Start gRPC Server (plugin): port=50051, max_workers=100

---
options: { }

---
options: { }
secret_data: { }