GitHub - PowerLoom/pooler at testnet_5_pairs

 "project_type": "pairContract_pair_total_reserves",
 "preload_tasks":[
   "eth_price",
   "block_details"
 ],
 "projects":[
   "0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc",
   "0xae461ca67b15dc8dc81ce7615e0320da1a9ab8d5",
   "0x0d4a11d5eeaac28ec3f61d100daf4d40471f1852",
   "0x3041cbd36888becc7bbcbc0045e3b1f144466f5f",
 class GenericPreloader(ABC):
     __metaclass__ = ABCMeta

     def __init__(self):
         pass

     @abstractmethod
     async def compute(
         self,
         epoch: EpochBase,
         redis_conn: aioredis.Redis,
         rpc_helper: RpcHelper,
     ):
         pass

     @abstractmethod
     async def cleanup(self):
         pass
 class GenericDelegatorPreloader(GenericPreloader):
     _epoch: EpochBase
     _channel: aio_pika.abc.AbstractChannel
     _exchange: aio_pika.abc.AbstractExchange
     _q_obj: aio_pika.abc.AbstractQueue
     _consumer_tag: str
     _redis_conn: aioredis.Redis
     _task_type: str
     _epoch_id: int
     _preload_successful_event: asyncio.Event
     _awaited_delegated_response_ids: set
     _collected_response_objects: Dict[int, Dict[str, Dict[Any, Any]]]
     _logger: loguru._logger.Logger
     _request_id_query_obj_map: Dict[int, Any]

     @abstractmethod
     async def _on_delegated_responses_complete(self):
         pass

     @abstractmethod
     async def _on_filter_worker_response_message(
         self,
         message: aio_pika.abc.AbstractIncomingMessage,
     ):
         pass

     @abstractmethod
     async def _handle_filter_worker_response_message(self, message_body: bytes):
         pass

     @abstractmethod
     async def _periodic_awaited_responses_checker(self):
         pass
 {
     "preloaders": [
         {
           "task_type": "block_transactions",
           "module": "snapshotter.utils.preloaders.tx_receipts.preloader",
           "class_name": "TxPreloadWorker"
         },
         {
           "task_type": "block_details",
           "module": "snapshotter.utils.preloaders.block_details.preloader",
           "class_name": "BlockDetailsPreloader"
         },
         {
           "task_type": "eth_price",
           "module": "snapshotter.utils.preloaders.eth_price.preloader",
           "class_name": "EthPricePreloader"
         }
     ],
     "delegate_tasks": [
         {
           "task_type": "txreceipt",
           "module": "snapshotter.utils.preloaders.tx_receipts.delegated_worker.tx_receipts",
           "class_name": "TxReceiptProcessor"
         }
     ],
     "timeout": 60
 }
 def _gen_project_id(self, type_: str, epoch):
     if not epoch.data_source:
         # For generic use cases that don't have a data source like block details
         project_id = f'{type_}:{settings.namespace}'
     else:
         if epoch.primary_data_source:
             project_id = f'{type_}:{epoch.primary_data_source}_{epoch.data_source}:{settings.namespace}'
         else:
             project_id = f'{type_}:{epoch.data_source}:{settings.namespace}'
     return project_id
 for project_type in self._project_type_config_mapping:
     project_config = self._project_type_config_mapping[project_type]
     if not project_config.preload_tasks:
         continue
     self._logger.debug(
         'Expected list of successful preloading for project type {}: {}',
         project_type,
         project_config.preload_tasks,
     )
     if all([t in succesful_preloads for t in project_config.preload_tasks]):
         self._logger.info(
             'Preloading dependency satisfied for project type {} epoch {}. Distributing snapshot build tasks...',
             project_type, epoch.epochId,
         )
         asyncio.ensure_future(
             self._redis_conn.hset(
                 name=epoch_id_project_to_state_mapping(epoch.epochId, SnapshotterStates.PRELOAD.value),
                 mapping={
                     project_type: SnapshotterStateUpdate(
                         status='success', timestamp=int(time.time())
                     ).json()
                 }
             )
         )
         await self._distribute_callbacks_snapshotting(project_type, epoch)
     await self._redis_conn.hset(
         name=epoch_id_project_to_state_mapping(
             epoch_id=msg_obj.epochId, state_id=SnapshotterStates.SNAPSHOT_BUILD.value,
         ),
         mapping={
             project_id: SnapshotterStateUpdate(
                 status='failed', error=str(e), timestamp=int(time.time()),
             ).json(),
         },
     )
 else:
     await self._redis_conn.hset(
         name=epoch_id_project_to_state_mapping(
             epoch_id=msg_obj.epochId, state_id=SnapshotterStates.SNAPSHOT_BUILD.value,
         ),
         mapping={
             project_id: SnapshotterStateUpdate(
                 status='success', timestamp=int(time.time()),
             ).json(),
         },
     )
 except Exception as e:
     self._logger.opt(exception=True).error(
         (
             'Exception committing snapshot to commit payload queue:'
             ' {} | dump: {}'
         ),
         snapshot,
         e,
     )
     await self._redis_conn.hset(
         name=epoch_id_project_to_state_mapping(
             epoch.epochId, SnapshotterStates.SNAPSHOT_SUBMIT_PAYLOAD_COMMIT.value,
         ),
         mapping={
             project_id: SnapshotterStateUpdate(
                 status='failed', error=str(e), timestamp=int(time.time()),
             ).json(),
         },
     )
 else:
     await self._redis_conn.hset(
         name=epoch_id_project_to_state_mapping(
             epoch.epochId, SnapshotterStates.SNAPSHOT_SUBMIT_PAYLOAD_COMMIT.value,
         ),
         mapping={
             project_id: SnapshotterStateUpdate(
                 status='success', timestamp=int(time.time()),
             ).json(),
         },
     )
 await self._redis_conn.hset(
     name=epoch_id_project_to_state_mapping(msg_obj.epochId, SnapshotterStates.SNAPSHOT_FINALIZE.value),
     mapping={
         msg_obj.projectId: SnapshotterStateUpdate(
             status='success', timestamp=int(time.time()), extra={'snapshot_cid': msg_obj.snapshotCid}
         ).json()
     }
 )
     for each_lambda in task_processor.transformation_lambdas:
         snapshot = each_lambda(snapshot, msg_obj)

 await self._send_payload_commit_service_queue(
     type_=task_type,
     project_id=project_id,
     epoch=msg_obj,
     snapshot=snapshot,
     storage_flag=settings.web3storage.upload_aggregates,
         await self._init_rpc_helper()
         await self._load_projects_metadata()
         await self._init_rabbitmq_connection()
         await self._init_preloader_compute_mapping()
         self._initialized = True

 async def _load_projects_metadata(self):
     if not self._projects_list:
         with open(settings.protocol_state.abi, 'r') as f:
             abi_dict = json.load(f)
         protocol_state_contract = self._anchor_rpc_helper.get_current_node()['web3_client'].eth.contract(
             address=Web3.toChecksumAddress(
                 settings.protocol_state.address,
             ),
             abi=abi_dict,
         )
         await get_source_chain_epoch_size(
         if epoch.epochId in self._preload_completion_conditions:
             del self._preload_completion_conditions[epoch.epochId]

     async def _exec_preloaders(
         self, msg_obj: EpochBase,
     ):
         """
         Functions to preload data points required by snapshot builders
         This is to save on redundant RPC and cache calls
         """
         # cleanup previous preloading complete tasks and events
         # start all preload tasks
         for preloader in preloaders:
             if preloader.task_type in self._all_preload_tasks:
                 preloader_class = self._preloader_compute_mapping[preloader.task_type]
                 preloader_obj = preloader_class()
                 preloader_compute_kwargs = dict(
                     epoch=msg_obj,
                     redis_conn=self._redis_conn,
                     rpc_helper=self._rpc_helper,
                 )
                 self._logger.debug(
                     'Starting preloader obj {} for epoch {}',
                     preloader.task_type,
                     msg_obj.epochId,
                 )
                 f = asyncio.ensure_future(preloader_obj.compute(**preloader_compute_kwargs))
                 self._preload_completion_conditions[msg_obj.epochId][preloader.task_type] = PreloaderAsyncFutureDetails(
                     obj=preloader_obj,
                     future=f,
                 )
         for project_config in self.projects_config:
             if not project_config.preload_tasks:
                 # release for snapshotting
                 asyncio.ensure_future(
                     self._distribute_callbacks_snapshotting(
                         project_config.project_type, msg_obj,
                     ),
                 )
                 continue

         asyncio.ensure_future(
             self._preloader_waiter(
                 epoch=msg_obj,
             ),
         )

     async def _epoch_release_processor(self, message: IncomingMessage):
         try:
             msg_obj: EpochBase = (
                 EpochBase.parse_raw(message.body)
             )
         except ValidationError:
             self._logger.opt(exception=True).error(
                 'Bad message structure of epoch callback',
             )
             return
         except Exception:
             self._logger.opt(exception=True).error(
                 'Unexpected message format of epoch callback',
             )
             return

         self._newly_added_projects = self._newly_added_projects.union(
             await self._enable_pending_projects_for_epoch(msg_obj.epochId),
         )

         await self._exec_preloaders(msg_obj=msg_obj)

     async def _distribute_callbacks_snapshotting(self, project_type: str, epoch: EpochBase):
         # send to snapshotters to get the balances of the addresses
         queuing_tasks = []

         async with self._rmq_channel_pool.acquire() as ch:
             # Prepare a message to send
             exchange = await ch.get_exchange(
 "delegate_tasks": [
     {
       "task_type": "txreceipt",
       "module": "snapshotter.utils.preloaders.tx_receipts.delegated_worker.tx_receipts",
       "class_name": "TxReceiptProcessor"
     }
 ],
 "callback_worker_config": {
   "num_delegate_workers": 8,
   "num_snapshot_workers": 2,
   "num_aggregation_workers": 4
 },
 "callbacks": {
   "exchange": "powerloom-backend-callbacks"
 },
     ch: pika.adapters.blocking_connection.BlockingChannel,
 ) -> None:
     callback_exchange_name = (
         f'{settings.rabbitmq.setup.callbacks.exchange}:{settings.namespace}'
     )
     # Snapshot queue
     queue_name, routing_key_pattern = get_snapshot_queue_routing_key_pattern()
     init_topic_exchange_and_queue(
         ch,
         exchange_name=callback_exchange_name,
         queue_name=queue_name,
         routing_key_pattern=routing_key_pattern,
     )

     # Aggregate queue
     queue_name, routing_key_pattern = get_aggregate_queue_routing_key_pattern()
     init_queue(
         ch,
         exchange_name=callback_exchange_name,
         queue_name=queue_name,
         routing_key=routing_key_pattern,
     )
 @app.get('/last_finalized_epoch/{project_id}')
 async def get_project_last_finalized_epoch_info(
     request: Request,
     response: Response,
     project_id: str,
     rate_limit_auth_dep: RateLimitAuthCheck = Depends(
         rate_limit_auth_check,
     ),
 ):
     """
     This endpoint is used to fetch epoch info for the last finalized epoch for a given project.
     """
     if not (
         rate_limit_auth_dep.rate_limit_passed and
         rate_limit_auth_dep.authorized and
         rate_limit_auth_dep.owner.active == UserStatusEnum.active
     ):
         return inject_rate_limit_fail_response(rate_limit_auth_dep)

     try:

         # get project last finalized epoch from redis
         project_last_finalized_epoch = await request.app.state.redis_pool.get(
             project_last_finalized_epoch_key(project_id),
         )

         if project_last_finalized_epoch is None:
             # find from contract
             epoch_finalized = False
             [cur_epoch] = await request.app.state.anchor_rpc_helper.web3_call(
                 [request.app.state.protocol_state_contract.functions.currentEpoch()],
                 redis_conn=request.app.state.redis_pool,
             )
             epoch_id = int(cur_epoch[2])
             while not epoch_finalized and epoch_id >= 0:
                 # get finalization status
                 [epoch_finalized_contract] = await request.app.state.anchor_rpc_helper.web3_call(
                     [request.app.state.protocol_state_contract.functions.snapshotStatus(project_id, epoch_id)],
                     redis_conn=request.app.state.redis_pool,
                 )
                 if epoch_finalized_contract[0]:
                     epoch_finalized = True
                     project_last_finalized_epoch = epoch_id
                     await request.app.state.redis_pool.set(
                         project_last_finalized_epoch_key(project_id),
                         project_last_finalized_epoch,
                     )
                 else:
                     epoch_id -= 1
                     if epoch_id < 0:
                         response.status_code = 404
                         return {
                             'status': 'error',
                             'message': f'Unable to find last finalized epoch for project {project_id}',
                         }
         else:
             project_last_finalized_epoch = int(project_last_finalized_epoch.decode('utf-8'))
         [epoch_info_data] = await request.app.state.anchor_rpc_helper.web3_call(
             [request.app.state.protocol_state_contract.functions.epochInfo(project_last_finalized_epoch)],
             redis_conn=request.app.state.redis_pool,
         )
         epoch_info = {
             'epochId': project_last_finalized_epoch,
             'timestamp': epoch_info_data[0],
             'blocknumber': epoch_info_data[1],
             'epochEnd': epoch_info_data[2],
         }

     except Exception as e:
         rest_logger.exception(
             'Exception in get_project_last_finalized_epoch_info',
             e=e,
         )
         response.status_code = 500
         return {
             'status': 'error',
             'message': f'Unable to get last finalized epoch for project {project_id}, error: {e}',
         }

     auth_redis_conn: aioredis.Redis = request.app.state.auth_aioredis_pool
     await incr_success_calls_count(auth_redis_conn, rate_limit_auth_dep)

     return epoch_info
 @app.get('/data/{epoch_id}/{project_id}/')
 async def get_data_for_project_id_epoch_id(
     request: Request,
     response: Response,
     project_id: str,
     epoch_id: int,
     rate_limit_auth_dep: RateLimitAuthCheck = Depends(
         rate_limit_auth_check,
     ),
 ):
     """
     This endpoint is used to fetch data for a given project_id and epoch_id.
     """
     if not (
         rate_limit_auth_dep.rate_limit_passed and
         rate_limit_auth_dep.authorized and
         rate_limit_auth_dep.owner.active == UserStatusEnum.active
     ):
         return inject_rate_limit_fail_response(rate_limit_auth_dep)

     try:
         data = await get_project_epoch_snapshot(
             request.app.state.redis_pool,
             request.app.state.protocol_state_contract,
             request.app.state.anchor_rpc_helper,
             request.app.state.ipfs_reader_client,
             epoch_id,
             project_id,
         )
     except Exception as e:
         rest_logger.exception(
             'Exception in get_data_for_project_id_epoch_id',
             e=e,
         )
         response.status_code = 500
         return {
             'status': 'error',
             'message': f'Unable to get data for project_id: {project_id},'
             f' epoch_id: {epoch_id}, error: {e}',
         }

     if not data:
         response.status_code = 404
         return {
             'status': 'error',
             'message': f'No data found for project_id: {project_id},'
             f' epoch_id: {epoch_id}',
         }
     auth_redis_conn: aioredis.Redis = request.app.state.auth_aioredis_pool
     await incr_success_calls_count(auth_redis_conn, rate_limit_auth_dep)

     return data
 {
   "config": [{
       "project_type": "uniswap_pairContract_pair_total_reserves",
       "projects":[
         "0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc",
         "0xae461ca67b15dc8dc81ce7615e0320da1a9ab8d5",
         "0x0d4a11d5eeaac28ec3f61d100daf4d40471f1852",
         "0x3041cbd36888becc7bbcbc0045e3b1f144466f5f",
         "0xd3d2e2692501a5c9ca623199d38826e513033a17",
         "0xbb2b8038a1640196fbe3e38816f3e67cba72d940",
         "0xa478c2975ab1ea89e8196811f51a7b7ade33eb11"
         ],
       "processor":{
         "module": "pooler.modules.uniswapv2.pair_total_reserves",
         "class_name": "PairTotalReservesProcessor"
       }
     },
     {
       "project_type": "uniswap_pairContract_trade_volume",
       "projects":[
         "0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc",
         "0xae461ca67b15dc8dc81ce7615e0320da1a9ab8d5",
         "0x0d4a11d5eeaac28ec3f61d100daf4d40471f1852",
         "0x3041cbd36888becc7bbcbc0045e3b1f144466f5f",
         "0xd3d2e2692501a5c9ca623199d38826e513033a17",
         "0xbb2b8038a1640196fbe3e38816f3e67cba72d940",
         "0xa478c2975ab1ea89e8196811f51a7b7ade33eb11"
         ],
         "processor":{
           "module": "pooler.modules.uniswapv2.trade_volume",
           "class_name": "TradeVolumeProcessor"
         }
     }
   ]
 }
 async def get_project_epoch_snapshot(
     redis_conn: aioredis.Redis, state_contract_obj, rpc_helper, ipfs_reader, epoch_id, project_id,
 ) -> dict:
     cid = await get_project_finalized_cid(redis_conn, state_contract_obj, rpc_helper, epoch_id, project_id)
     if cid:
         data = await get_submission_data(redis_conn, cid, ipfs_reader, project_id)
         return data
     else:
         return dict()
 async def get_tail_epoch_id(
         redis_conn: aioredis.Redis,
         state_contract_obj,
         rpc_helper,
         current_epoch_id,
         time_in_seconds,
         project_id,
 ):
     # Returns tail epoch_id and a boolean indicating if tail contains the full time window
     source_chain_epoch_size = await get_source_chain_epoch_size(redis_conn, state_contract_obj, rpc_helper)
     source_chain_block_time = await get_source_chain_block_time(redis_conn, state_contract_obj, rpc_helper)

     # calculate tail epoch_id
     tail_epoch_id = current_epoch_id - int(time_in_seconds / (source_chain_epoch_size * source_chain_block_time))
     project_first_epoch = await(
         get_project_first_epoch(redis_conn, state_contract_obj, rpc_helper, project_id)
     )
     if tail_epoch_id < project_first_epoch:
         tail_epoch_id = project_first_epoch
         return tail_epoch_id, True

     logger.trace(
         'project ID {} tail epoch_id: {} against head epoch ID {} ',
         project_id, tail_epoch_id, current_epoch_id,
     )

     return tail_epoch_id, False
 "project_type": "aggregate_uniswap_pairContract_24h_trade_volume",
 "aggregate_on": "SingleProject",
 "filters": {
   "projectId": "uniswap_pairContract_trade_volume"
 },
 "processor": {
   "module": "pooler.modules.uniswapv2.aggregate.single_uniswap_trade_volume_24h",
   "class_name": "AggreagateTradeVolumeProcessor"
Name		Name	Last commit message	Last commit date
Latest commit History 1,021 Commits
.github		.github
config		config
logs		logs
snapshotter		snapshotter
.editorconfig		.editorconfig
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
.pylintrc		.pylintrc
Dockerfile		Dockerfile
LICENSE		LICENSE
README.md		README.md
build-docker.sh		build-docker.sh
dev-requirements.txt		dev-requirements.txt
init_processes.sh		init_processes.sh
mypy.ini		mypy.ini
pm2.config.js		pm2.config.js
poetry.lock		poetry.lock
pyproject.toml		pyproject.toml
setup.cfg		setup.cfg
snapshotter_autofill.sh		snapshotter_autofill.sh
	"project_type": "pairContract_pair_total_reserves",
	"preload_tasks":[
	"eth_price",
	"block_details"
	],
	"projects":[
	"0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc",
	"0xae461ca67b15dc8dc81ce7615e0320da1a9ab8d5",
	"0x0d4a11d5eeaac28ec3f61d100daf4d40471f1852",
	"0x3041cbd36888becc7bbcbc0045e3b1f144466f5f",
	class GenericPreloader(ABC):
	__metaclass__ = ABCMeta

	def __init__(self):
	pass

	@abstractmethod
	async def compute(
	self,
	epoch: EpochBase,
	redis_conn: aioredis.Redis,
	rpc_helper: RpcHelper,
	):
	pass

	@abstractmethod
	async def cleanup(self):
	pass
	class GenericDelegatorPreloader(GenericPreloader):
	_epoch: EpochBase
	_channel: aio_pika.abc.AbstractChannel
	_exchange: aio_pika.abc.AbstractExchange
	_q_obj: aio_pika.abc.AbstractQueue
	_consumer_tag: str
	_redis_conn: aioredis.Redis
	_task_type: str
	_epoch_id: int
	_preload_successful_event: asyncio.Event
	_awaited_delegated_response_ids: set
	_collected_response_objects: Dict[int, Dict[str, Dict[Any, Any]]]
	_logger: loguru._logger.Logger
	_request_id_query_obj_map: Dict[int, Any]

	@abstractmethod
	async def _on_delegated_responses_complete(self):
	pass

	@abstractmethod
	async def _on_filter_worker_response_message(
	self,
	message: aio_pika.abc.AbstractIncomingMessage,
	):
	pass

	@abstractmethod
	async def _handle_filter_worker_response_message(self, message_body: bytes):
	pass

	@abstractmethod
	async def _periodic_awaited_responses_checker(self):
	pass
	{
	"preloaders": [
	{
	"task_type": "block_transactions",
	"module": "snapshotter.utils.preloaders.tx_receipts.preloader",
	"class_name": "TxPreloadWorker"
	},
	{
	"task_type": "block_details",
	"module": "snapshotter.utils.preloaders.block_details.preloader",
	"class_name": "BlockDetailsPreloader"
	},
	{
	"task_type": "eth_price",
	"module": "snapshotter.utils.preloaders.eth_price.preloader",
	"class_name": "EthPricePreloader"
	}
	],
	"delegate_tasks": [
	{
	"task_type": "txreceipt",
	"module": "snapshotter.utils.preloaders.tx_receipts.delegated_worker.tx_receipts",
	"class_name": "TxReceiptProcessor"
	}
	],
	"timeout": 60
	}
	def _gen_project_id(self, type_: str, epoch):
	if not epoch.data_source:
	# For generic use cases that don't have a data source like block details
	project_id = f'{type_}:{settings.namespace}'
	else:
	if epoch.primary_data_source:
	project_id = f'{type_}:{epoch.primary_data_source}_{epoch.data_source}:{settings.namespace}'
	else:
	project_id = f'{type_}:{epoch.data_source}:{settings.namespace}'
	return project_id
	for project_type in self._project_type_config_mapping:
	project_config = self._project_type_config_mapping[project_type]
	if not project_config.preload_tasks:
	continue
	self._logger.debug(
	'Expected list of successful preloading for project type {}: {}',
	project_type,
	project_config.preload_tasks,
	)
	if all([t in succesful_preloads for t in project_config.preload_tasks]):
	self._logger.info(
	'Preloading dependency satisfied for project type {} epoch {}. Distributing snapshot build tasks...',
	project_type, epoch.epochId,
	)
	asyncio.ensure_future(
	self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(epoch.epochId, SnapshotterStates.PRELOAD.value),
	mapping={
	project_type: SnapshotterStateUpdate(
	status='success', timestamp=int(time.time())
	).json()
	}
	)
	)
	await self._distribute_callbacks_snapshotting(project_type, epoch)
	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(
	epoch_id=msg_obj.epochId, state_id=SnapshotterStates.SNAPSHOT_BUILD.value,
	),
	mapping={
	project_id: SnapshotterStateUpdate(
	status='failed', error=str(e), timestamp=int(time.time()),
	).json(),
	},
	)
	else:
	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(
	epoch_id=msg_obj.epochId, state_id=SnapshotterStates.SNAPSHOT_BUILD.value,
	),
	mapping={
	project_id: SnapshotterStateUpdate(
	status='success', timestamp=int(time.time()),
	).json(),
	},
	)
	except Exception as e:
	self._logger.opt(exception=True).error(
	(
	'Exception committing snapshot to commit payload queue:'
	' {} \| dump: {}'
	),
	snapshot,
	e,
	)
	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(
	epoch.epochId, SnapshotterStates.SNAPSHOT_SUBMIT_PAYLOAD_COMMIT.value,
	),
	mapping={
	project_id: SnapshotterStateUpdate(
	status='failed', error=str(e), timestamp=int(time.time()),
	).json(),
	},
	)
	else:
	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(
	epoch.epochId, SnapshotterStates.SNAPSHOT_SUBMIT_PAYLOAD_COMMIT.value,
	),
	mapping={
	project_id: SnapshotterStateUpdate(
	status='success', timestamp=int(time.time()),
	).json(),
	},
	)
	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(msg_obj.epochId, SnapshotterStates.SNAPSHOT_FINALIZE.value),
	mapping={
	msg_obj.projectId: SnapshotterStateUpdate(
	status='success', timestamp=int(time.time()), extra={'snapshot_cid': msg_obj.snapshotCid}
	).json()
	}
	)