通过django提高python 编码(一)

通过django来进阶python,看看django对常用问题的写法,提升自己。

本文方法来之django/core/cache/backends/filebased.py

• 字典先取key1的value,如果key不存在，则取key2的value,如果都不在取默认值

  // params 字典
  timeout = params.get('timeout', params.get('TIMEOUT', 300))
  

• 用户传递的函数对象或函数名都能执行

  def get_key_func(key_func):
      if key_func is not None:
          if callable(key_func):
              return key_func
          else:
              return import_string(key_func)
      return default_key_func
  

• 验证可移植的memcached秘钥

  MEMCACHE_MAX_KEY_LENGTH = 250
  
  def validate_key(self, key):
      if len(key) > MEMCACHE_MAX_KEY_LENGTH:
          warnings.warn(
              'Cache key will cause errors if used with memcached: %r '
              '(longer than %s)' % (key, MEMCACHE_MAX_KEY_LENGTH), CacheKeyWarning
          )
      for char in key:
          if ord(char) < 33 or ord(char) == 127:
              warnings.warn(
                  'Cache key contains characters that will cause errors if '
                  'used with memcached: %r' % key, CacheKeyWarning
              )
              break
  

• 创建目录

      def _createdir(self):
          if not os.path.exists(self._dir):
              try:
                  os.makedirs(self._dir, 0o700)
              except FileExistsError:
                  pass
  

• 列出目录下的文件夹

  # 自己改了下
  import os
  import glob
  
  dir_ = os.path.abspath('./')
  cache_suffix = '.py'
  
  def _list_cache_files(_dir):
      """
      Get a list of paths to all the cache files. These are all the files
      in the root cache dir that end on the cache_suffix.
      """
      if not os.path.exists(_dir):
          return []
      filelist = [os.path.join(_dir, fname) for fname
                  in glob.glob1(_dir, '*%s' % cache_suffix)]
      return filelist
  
  print(_list_cache_files(dir_))
  

• 删除文件

  def _delete(self, fname):
      if not fname.startswith(self._dir) or not os.path.exists(fname):
          return
      try:
          os.remove(fname)
      except FileNotFoundError:
          # The file may have been removed by another process.
          pass
  
  

• 当缓存满的时候，移除一部分缓存算法

  _max_entries = 300
  _cull_frequency = 3
  def _cull():
  
      """
      清理缓存的触发点是超过做大缓存数目
      随机清除缓存三分之一
      _cull_frequency 为0 表示清空
      """
      filelist = _list_cache_files()
      num_entries = len(filelist)
      if num_entries < _max_entries:
          return  # return early if no culling is required
      if _cull_frequency == 0:
          return self.clear()  # Clear the cache when CULL_FREQUENCY = 0
      # Delete a random selection of entries
      filelist = random.sample(filelist,
                               int(num_entries / self._cull_frequency))
      for fname in filelist:
          self._delete(fname)
  

• 用文件作为缓存后端，设置值的逻辑

  
  def set(self, key, value, timeout=DEFAULT_TIMEOUT, version=None):
      """
          先检查缓存文件所在文件夹在不在
          生成缓存文件的目标名字
          确认是否清理缓存
          生成中间文件，内容写入中间文件，移动中间文件到目标文件
          如果失败移除中间文件
      """
      self._createdir()  # Cache dir can be deleted at any time.
      fname = self._key_to_file(key, version)
      self._cull()  # make some room if necessary
      fd, tmp_path = tempfile.mkstemp(dir=self._dir)
      renamed = False
      try:
          with open(fd, 'wb') as f:
              self._write_content(f, timeout, value)
          file_move_safe(tmp_path, fname, allow_overwrite=True)
          renamed = True
      finally:
          if not renamed:
              os.remove(tmp_path)
  

本文方法来之django/core/files/move.py

• 移动或拷贝文件兼容win和unix

  def _samefile(src, dst):
      # Mac, Unix.使用自带的os.path.samefile
      if hasattr(os.path, 'samefile'):
          try:
              return os.path.samefile(src, dst)
          except OSError:
              return False
  
      # 其他系统，检查路径
      return (os.path.normcase(os.path.abspath(src)) ==
              os.path.normcase(os.path.abspath(dst)))
  
  def file_move_safe(old_file_name, new_file_name, chunk_size=1024 * 64, allow_overwrite=False):
      """
      以安全的方式移动文件，首先使用 os.rename移动，如果失败，手动拷贝。
      如果目标文件存在取决于 allow_overwrite 的值
      拷贝到时候加锁，这个锁也是跨平台
      """
      # 如果文件名一样，直接返回
      if _samefile(old_file_name, new_file_name):
          return
  
      try:
          # os.access 用于检测权限以及路径是否存在
          if not allow_overwrite and os.access(new_file_name, os.F_OK):
              raise IOError("Destination file %s exists and allow_overwrite is False" % new_file_name)
          # 以便宜的方式拷贝
          os.rename(old_file_name, new_file_name)
          return
      except OSError:
          # OSError发生有两种情况：
          #     移动到不同的文件系统上（ext3,ext4,tffs）
          #     移动打开文件
          pass
  
      # first open the old file, so that it won't go away
      with open(old_file_name, 'rb') as old_file:
          # 此处使用os打开文件标准做法（O_BINARY window必须）：
          fd = os.open(new_file_name, (os.O_WRONLY | os.O_CREAT | getattr(os, 'O_BINARY', 0) |
                                      (os.O_EXCL if not allow_overwrite else 0)))
          try:
              # 上锁
              locks.lock(fd, locks.LOCK_EX)
              current_chunk = None
              while current_chunk != b'':
                  current_chunk = old_file.read(chunk_size)
                  os.write(fd, current_chunk)
          finally:
              # 释放锁，关闭文件句柄
              locks.unlock(fd)
              os.close(fd)
  
      try:
          # 复制权限
          copystat(old_file_name, new_file_name)
      except PermissionError as e:
          # 某些文件系统在拷贝文件元数据的时候会失败（CIFS）,
          # 如果old_file_name, new_file_name 两个文件系统不同，忽略
          if e.errno != errno.EPERM:
              raise
  
      try:
          os.remove(old_file_name)
      except PermissionError as e:
          # 当删除文件的时候，某些操作系统（Cygwin and Windows)）会失败，直接忽略就行。因为临时文件会自动删除
          if getattr(e, 'winerror', 0) != 32:
              raise