FM / persisted

/usr/lib/python3/dist-packages/twisted/persisted/

/usr/lib/python3/dist-packages/twisted/persisted

Name	Size	Mode	Actions
test/	-	0755	rm
__pycache__/	-	0755	rm
aot.py	18605	0644	edit dl rm
crefutil.py	4375	0644	edit dl rm
dirdbm.py	10141	0644	edit dl rm
sob.py	5197	0644	edit dl rm
styles.py	13495	0644	edit dl rm
__init__.py	136	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/persisted/dirdbm.py (10141B)

# -*- test-case-name: twisted.test.test_dirdbm -*-
#
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
DBM-style interface to a directory.

Each key is stored as a single file.  This is not expected to be very fast or
efficient, but it's good for easy debugging.

DirDBMs are *not* thread-safe, they should only be accessed by one thread at
a time.

No files should be placed in the working directory of a DirDBM save those
created by the DirDBM itself!

Maintainer: Itamar Shtull-Trauring
"""

import os
import base64
import glob

try:
    import cPickle as pickle
except ImportError:
    import pickle

from twisted.python.filepath import FilePath

try:
    _open
except NameError:
    _open = open

class DirDBM:
    """
    A directory with a DBM interface.

This class presents a hash-like interface to a directory of small,
    flat files. It can only use strings as keys or values.
    """

def __init__(self, name):
        """
        @type name: str
        @param name: Base path to use for the directory storage.
        """
        self.dname = os.path.abspath(name)
        self._dnamePath = FilePath(name)
        if not self._dnamePath.isdir():
            self._dnamePath.createDirectory()
        else:
            # Run recovery, in case we crashed. we delete all files ending
            # with ".new". Then we find all files who end with ".rpl". If a
            # corresponding file exists without ".rpl", we assume the write
            # failed and delete the ".rpl" file. If only a ".rpl" exist we
            # assume the program crashed right after deleting the old entry
            # but before renaming the replacement entry.
            #
            # NOTE: '.' is NOT in the base64 alphabet!
            for f in glob.glob(self._dnamePath.child("*.new").path):
                os.remove(f)
            replacements = glob.glob(self._dnamePath.child("*.rpl").path)
            for f in replacements:
                old = f[:-4]
                if os.path.exists(old):
                    os.remove(f)
                else:
                    os.rename(f, old)

def _encode(self, k):
        """
        Encode a key so it can be used as a filename.
        """
        # NOTE: '_' is NOT in the base64 alphabet!
        return base64.encodestring(k).replace(b'\n', b'_').replace(b"/", b"-")

def _decode(self, k):
        """
        Decode a filename to get the key.
        """
        return base64.decodestring(k.replace(b'_', b'\n').replace(b"-", b"/"))

def _readFile(self, path):
        """
        Read in the contents of a file.

Override in subclasses to e.g. provide transparently encrypted dirdbm.
        """
        with _open(path.path, "rb") as f:
            s = f.read()
        return s

def _writeFile(self, path, data):
        """
        Write data to a file.

Override in subclasses to e.g. provide transparently encrypted dirdbm.
        """
        with _open(path.path, "wb") as f:
            f.write(data)
            f.flush()

def __len__(self):
        """
        @return: The number of key/value pairs in this Shelf
        """
        return len(self._dnamePath.listdir())

def __setitem__(self, k, v):
        """
        C{dirdbm[k] = v}
        Create or modify a textfile in this directory

@type k: bytes
        @param k: key to set

@type v: bytes
        @param v: value to associate with C{k}
        """
        if not type(k) == bytes:
            raise TypeError("DirDBM key must be bytes")
        if not type(v) == bytes:
            raise TypeError("DirDBM value must be bytes")
        k = self._encode(k)

# We create a new file with extension .new, write the data to it, and
        # if the write succeeds delete the old file and rename the new one.
        old = self._dnamePath.child(k)
        if old.exists():
            new = old.siblingExtension(".rpl") # Replacement entry
        else:
            new = old.siblingExtension(".new") # New entry
        try:
            self._writeFile(new, v)
        except:
            new.remove()
            raise
        else:
            if (old.exists()): old.remove()
            new.moveTo(old)

def __getitem__(self, k):
        """
        C{dirdbm[k]}
        Get the contents of a file in this directory as a string.

@type k: bytes
        @param k: key to lookup

@return: The value associated with C{k}
        @raise KeyError: Raised when there is no such key
        """
        if not type(k) == bytes:
            raise TypeError("DirDBM key must be bytes")
        path = self._dnamePath.child(self._encode(k))
        try:
            return self._readFile(path)
        except (EnvironmentError):
            raise KeyError(k)

def __delitem__(self, k):
        """
        C{del dirdbm[foo]}
        Delete a file in this directory.

@type k: bytes
        @param k: key to delete

@raise KeyError: Raised when there is no such key
        """
        if not type(k) == bytes:
            raise TypeError("DirDBM key must be bytes")
        k = self._encode(k)
        try:
            self._dnamePath.child(k).remove()
        except (EnvironmentError):
            raise KeyError(self._decode(k))

def keys(self):
        """
        @return: a L{list} of filenames (keys).
        """
        return list(map(self._decode, self._dnamePath.asBytesMode().listdir()))

def values(self):
        """
        @return: a L{list} of file-contents (values).
        """
        vals = []
        keys = self.keys()
        for key in keys:
            vals.append(self[key])
        return vals

def items(self):
        """
        @return: a L{list} of 2-tuples containing key/value pairs.
        """
        items = []
        keys = self.keys()
        for key in keys:
            items.append((key, self[key]))
        return items

def has_key(self, key):
        """
        @type key: bytes
        @param key: The key to test

@return: A true value if this dirdbm has the specified key, a false
        value otherwise.
        """
        if not type(key) == bytes:
            raise TypeError("DirDBM key must be bytes")
        key = self._encode(key)
        return self._dnamePath.child(key).isfile()

def setdefault(self, key, value):
        """
        @type key: bytes
        @param key: The key to lookup

@param value: The value to associate with key if key is not already
        associated with a value.
        """
        if key not in self:
            self[key] = value
            return value
        return self[key]

def get(self, key, default = None):
        """
        @type key: bytes
        @param key: The key to lookup

@param default: The value to return if the given key does not exist

@return: The value associated with C{key} or C{default} if not
        L{DirDBM.has_key(key)}
        """
        if key in self:
            return self[key]
        else:
            return default

def __contains__(self, key):
        """
        @see: L{DirDBM.has_key}
        """
        return self.has_key(key)

def update(self, dict):
        """
        Add all the key/value pairs in L{dict} to this dirdbm.  Any conflicting
        keys will be overwritten with the values from L{dict}.

@type dict: mapping
        @param dict: A mapping of key/value pairs to add to this dirdbm.
        """
        for key, val in dict.items():
            self[key]=val

def copyTo(self, path):
        """
        Copy the contents of this dirdbm to the dirdbm at C{path}.

@type path: L{str}
        @param path: The path of the dirdbm to copy to.  If a dirdbm
        exists at the destination path, it is cleared first.

@rtype: C{DirDBM}
        @return: The dirdbm this dirdbm was copied to.
        """
        path = FilePath(path)
        assert path != self._dnamePath

d = self.__class__(path.path)
        d.clear()
        for k in self.keys():
            d[k] = self[k]
        return d

def clear(self):
        """
        Delete all key/value pairs in this dirdbm.
        """
        for k in self.keys():
            del self[k]

def close(self):
        """
        Close this dbm: no-op, for dbm-style interface compliance.
        """

def getModificationTime(self, key):
        """
        Returns modification time of an entry.

@return: Last modification date (seconds since epoch) of entry C{key}
        @raise KeyError: Raised when there is no such key
        """
        if not type(key) == bytes:
            raise TypeError("DirDBM key must be bytes")
        path = self._dnamePath.child(self._encode(key))
        if path.isfile():
            return path.getModificationTime()
        else:
            raise KeyError(key)

class Shelf(DirDBM):
    """
    A directory with a DBM shelf interface.

This class presents a hash-like interface to a directory of small,
    flat files. Keys must be strings, but values can be any given object.
    """

def __setitem__(self, k, v):
        """
        C{shelf[foo] = bar}
        Create or modify a textfile in this directory.

@type k: str
        @param k: The key to set

@param v: The value to associate with C{key}
        """
        v = pickle.dumps(v)
        DirDBM.__setitem__(self, k, v)

def __getitem__(self, k):
        """
        C{dirdbm[foo]}
        Get and unpickle the contents of a file in this directory.

@type k: bytes
        @param k: The key to lookup

@return: The value associated with the given key
        @raise KeyError: Raised if the given key does not exist
        """
        return pickle.loads(DirDBM.__getitem__(self, k))

def open(file, flag = None, mode = None):
    """
    This is for 'anydbm' compatibility.

@param file: The parameter to pass to the DirDBM constructor.

@param flag: ignored
    @param mode: ignored
    """
    return DirDBM(file)

__all__ = ["open", "DirDBM", "Shelf"]

cmd: