| |
| from pathlib import Path |
| from typing import Union |
|
|
| from .base import BaseStorageBackend |
|
|
|
|
| class MemcachedBackend(BaseStorageBackend): |
| """Memcached storage backend. |
| |
| Attributes: |
| server_list_cfg (str): Config file for memcached server list. |
| client_cfg (str): Config file for memcached client. |
| sys_path (str, optional): Additional path to be appended to `sys.path`. |
| Defaults to None. |
| """ |
|
|
| def __init__(self, server_list_cfg, client_cfg, sys_path=None): |
| if sys_path is not None: |
| import sys |
| sys.path.append(sys_path) |
| try: |
| import mc |
| except ImportError: |
| raise ImportError( |
| 'Please install memcached to enable MemcachedBackend.') |
|
|
| self.server_list_cfg = server_list_cfg |
| self.client_cfg = client_cfg |
| self._client = mc.MemcachedClient.GetInstance(self.server_list_cfg, |
| self.client_cfg) |
| |
| self._mc_buffer = mc.pyvector() |
|
|
| def get(self, filepath: Union[str, Path]): |
| """Get values according to the filepath. |
| |
| Args: |
| filepath (str or Path): Path to read data. |
| |
| Returns: |
| bytes: Expected bytes object. |
| |
| Examples: |
| >>> server_list_cfg = '/path/of/server_list.conf' |
| >>> client_cfg = '/path/of/mc.conf' |
| >>> backend = MemcachedBackend(server_list_cfg, client_cfg) |
| >>> backend.get('/path/of/file') |
| b'hello world' |
| """ |
| filepath = str(filepath) |
| import mc |
| self._client.Get(filepath, self._mc_buffer) |
| value_buf = mc.ConvertBuffer(self._mc_buffer) |
| return value_buf |
|
|
| def get_text(self, filepath, encoding=None): |
| raise NotImplementedError |
|
|
