parameters: (self)
return_statement: return ret
docstring: Load the state of the threat_updates checkpoints from state directory
docstring_summary: Load the state of the threat_updates checkpoints from state directory
docstring_tokens: Load the state of the threat_updates checkpoints from state directory
function: def _load_checkpoint(self) -> tu.ThreatUpdateCheckpoint:
          """Load the state of the threat_updates checkpoints from state directory"""
          txt_content = read_s3_text(
              self.s3_client, self.s3_bucket_name, self.checkpoint_s3_key
          )
  
          if txt_content is None:
              logger.warning("No s3 checkpoint for %d. First run?", self.privacy_group)
              return tu.ThreatUpdateCheckpoint()
          checkpoint_json = json.load(txt_content)
  
          ret = tu.ThreatUpdateCheckpoint(
              checkpoint_json["last_fetch_time"],
              checkpoint_json["fetch_checkpoint"],
          )
          logger.info(
              "Loaded checkpoint for privacy group %d. last_fetch_time=%d fetch_checkpoint=%d",
              self.privacy_group,
              ret.last_fetch_time,
              ret.fetch_checkpoint,
          )
  
          return ret
function_tokens: def _load_checkpoint ( self ) -> tu . ThreatUpdateCheckpoint : txt_content = read_s3_text ( self . s3_client , self . s3_bucket_name , self . checkpoint_s3_key ) if txt_content is None : logger . warning ( "No s3 checkpoint for %d. First run?" , self . privacy_group ) return tu . ThreatUpdateCheckpoint ( ) checkpoint_json = json . load ( txt_content ) ret = tu . ThreatUpdateCheckpoint ( checkpoint_json [ "last_fetch_time" ] , checkpoint_json [ "fetch_checkpoint" ] , ) logger . info ( "Loaded checkpoint for privacy group %d. last_fetch_time=%d fetch_checkpoint=%d" , self . privacy_group , ret . last_fetch_time , ret . fetch_checkpoint , ) return ret