status-im
/
airbyte-custom-connector
mirror of https://github.com/status-im/airbyte-custom-connector.git
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

source-crypto-market-extractor: fix coin name not being correctly passed 

Signed-off-by: Alexis Pentori <alexis@status.im>
This commit is contained in:
Alexis Pentori 2024-01-16 16:11:38 +01:00
parent e1257ad5ff
commit ebf65b1b78
No known key found for this signature in database
GPG Key ID: 65250D2801E47A10
1 changed files with 6 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
source-crypto-market-extractor/source_crypto_market_extractor/source.py
View File

@ -27,11 +27,10 @@ class CoinPrice(HttpStream):
return None return None
def stream_slices(self, **kwargs) -> Iterable[Optional[Mapping[str, Any]]]: def stream_slices(self, **kwargs) -> Iterable[Optional[Mapping[str, Any]]]:
for coin in self.coins: for coin in self.coins:
yield { yield {
"coin": coin, "coin": coin
} }
def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str: def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
@ -43,21 +42,21 @@ class CoinPrice(HttpStream):
stream_slice: Mapping[str, Any] = None, stream_slice: Mapping[str, Any] = None,
**kwargs **kwargs
) -> Iterable[Mapping]: ) -> Iterable[Mapping]:
logger.info("Parsing Coin Gecko date for %s", stream_slice['coin']) coin=stream_slice["coin"]
logger.info("Parsing Coin Gecko data for %s", coin)
market_chart = response.json() market_chart = response.json()
yield { yield {
"coin": stream_slices['coin'], "coin": coin,
"date": datetime.today().strftime('%Y%m%d_%H%M'), "date": datetime.today().strftime('%Y%m%d_%H%M'),
# The first value of the prices objects is weird # The first value of the prices objects is weird
"price": market_chart['prices'][1][1], "price": market_chart['prices'][1][1],
} }
# Source # Source
class SourceCryptoMarketExtractor(AbstractSource): class SourceCryptoMarketExtractor(AbstractSource):
def check_connection(self, logger, config) -> Tuple[bool, any]: def check_connection(self, logger, config) -> Tuple[bool, any]:
return True, None return True, None
def streams(self, config: Mapping[str, Any]) -> List[Stream]: def streams(self, config: Mapping[str, Any]) -> List[Stream]:
logger.info('Config : %s', config['coins'])
return [CoinPrice(coins=config['coins'])] return [CoinPrice(coins=config['coins'])]