NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

version_disk_storage.py (1990B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\caching\client\storage\version_disk_storage.py
      2 from caching.client.storage.base_storage import BaseVersionedStorage
      3 import os
      4 from caching.client.storage.disk_utils import _get_path, _get_data_path
      5 from semantic_version import SemanticVersion
      6 from semantic_version.errors import FailedToParseVersion
      7 
      8 def _get_version_path(content_id):
      9     return _get_path(file_name='version', file_type='.txt', content_id=content_id)
     10 
     11 
     12 class VersionedBinaryDiskStorage(BaseVersionedStorage):
     13 
     14     def __init__(self, content_id):
     15         super(VersionedBinaryDiskStorage, self).__init__(content_id)
     16 
     17     def get(self):
     18         path = _get_data_path(self.content_id, file_type='.bin')
     19         if not os.path.exists(path):
     20             return
     21         with open(path, 'rb') as f:
     22             data = f.read()
     23             return data
     24 
     25     def set(self, value):
     26         with open(_get_data_path(self.content_id, file_type='.bin'), 'wb') as f:
     27             f.write(value)
     28 
     29     def get_version(self):
     30         version_path = _get_version_path(self.content_id)
     31         if os.path.exists(version_path):
     32             with open(version_path, 'r') as f:
     33                 version = f.read().strip()
     34             try:
     35                 semantic_version = SemanticVersion.from_string(version)
     36                 if semantic_version == SemanticVersion(0, 0, 0, [], []):
     37                     return
     38                 return semantic_version
     39             except (ValueError, FailedToParseVersion):
     40                 return
     41 
     42     def set_version(self, version):
     43         with open(_get_version_path(self.content_id), 'w') as f:
     44             f.write(str(version))
     45 
     46     def clear(self):
     47         data_path = _get_data_path(self.content_id, file_type='.bin')
     48         if os.path.exists(data_path):
     49             os.remove(data_path)
     50         version_path = _get_version_path(self.content_id)
     51         if os.path.exists(version_path):
     52             os.remove(version_path)