Lib/bsddb/dbrecio.py

   1
   2 """
   3 File-like objects that read from or write to a bsddb record.
   4
   5 This implements (nearly) all stdio methods.
   6
   7 f = DBRecIO(db, key, txn=None)
   8 f.close()           # explicitly release resources held
   9 flag = f.isatty()   # always false
  10 pos = f.tell()      # get current position
  11 f.seek(pos)         # set current position
  12 f.seek(pos, mode)   # mode 0: absolute; 1: relative; 2: relative to EOF
  13 buf = f.read()      # read until EOF
  14 buf = f.read(n)     # read up to n bytes
  15 f.truncate([size])  # truncate file at to at most size (default: current pos)
  16 f.write(buf)        # write at current position
  17 f.writelines(list)  # for line in list: f.write(line)
  18
  19 Notes:
  20 - fileno() is left unimplemented so that code which uses it triggers
  21   an exception early.
  22 - There's a simple test set (see end of this file) - not yet updated
  23   for DBRecIO.
  24 - readline() is not implemented yet.
  25
  26
  27 From:
  28     Itamar Shtull-Trauring <itamar@maxnm.com>
  29 """
  30
  31 import errno
  32 import string
  33
  34 class DBRecIO:
  35     def __init__(self, db, key, txn=None):
  36         self.db = db
  37         self.key = key
  38         self.txn = txn
  39         self.len = None
  40         self.pos = 0
  41         self.closed = 0
  42         self.softspace = 0
  43
  44     def close(self):
  45         if not self.closed:
  46             self.closed = 1
  47             del self.db, self.txn
  48
  49     def isatty(self):
  50         if self.closed:
  51             raise ValueError, "I/O operation on closed file"
  52         return 0
  53
  54     def seek(self, pos, mode = 0):
  55         if self.closed:
  56             raise ValueError, "I/O operation on closed file"
  57         if mode == 1:
  58             pos = pos + self.pos
  59         elif mode == 2:
  60             pos = pos + self.len
  61         self.pos = max(0, pos)
  62
  63     def tell(self):
  64         if self.closed:
  65             raise ValueError, "I/O operation on closed file"
  66         return self.pos
  67
  68     def read(self, n = -1):
  69         if self.closed:
  70             raise ValueError, "I/O operation on closed file"
  71         if n < 0:
  72             newpos = self.len
  73         else:
  74             newpos = min(self.pos+n, self.len)
  75
  76         dlen = newpos - self.pos
  77
  78         r = self.db.get(key, txn=self.txn, dlen=dlen, doff=self.pos)
  79         self.pos = newpos
  80         return r
  81
  82     __fixme = """
  83     def readline(self, length=None):
  84         if self.closed:
  85             raise ValueError, "I/O operation on closed file"
  86         if self.buflist:
  87             self.buf = self.buf + string.joinfields(self.buflist, '')
  88             self.buflist = []
  89         i = string.find(self.buf, '\n', self.pos)
  90         if i < 0:
  91             newpos = self.len
  92         else:
  93             newpos = i+1
  94         if length is not None:
  95             if self.pos + length < newpos:
  96                 newpos = self.pos + length
  97         r = self.buf[self.pos:newpos]
  98         self.pos = newpos
  99         return r
 100
 101     def readlines(self, sizehint = 0):
 102         total = 0
 103         lines = []
 104         line = self.readline()
 105         while line:
 106             lines.append(line)
 107             total += len(line)
 108             if 0 < sizehint <= total:
 109                 break
 110             line = self.readline()
 111         return lines
 112     """
 113
 114     def truncate(self, size=None):
 115         if self.closed:
 116             raise ValueError, "I/O operation on closed file"
 117         if size is None:
 118             size = self.pos
 119         elif size < 0:
 120             raise IOError(errno.EINVAL,
 121                                       "Negative size not allowed")
 122         elif size < self.pos:
 123             self.pos = size
 124         self.db.put(key, "", txn=self.txn, dlen=self.len-size, doff=size)
 125
 126     def write(self, s):
 127         if self.closed:
 128             raise ValueError, "I/O operation on closed file"
 129         if not s: return
 130         if self.pos > self.len:
 131             self.buflist.append('\0'*(self.pos - self.len))
 132             self.len = self.pos
 133         newpos = self.pos + len(s)
 134         self.db.put(key, s, txn=self.txn, dlen=len(s), doff=self.pos)
 135         self.pos = newpos
 136
 137     def writelines(self, list):
 138         self.write(string.joinfields(list, ''))
 139
 140     def flush(self):
 141         if self.closed:
 142             raise ValueError, "I/O operation on closed file"
 143
 144
 145 """
 146 # A little test suite
 147
 148 def _test():
 149     import sys
 150     if sys.argv[1:]:
 151         file = sys.argv[1]
 152     else:
 153         file = '/etc/passwd'
 154     lines = open(file, 'r').readlines()
 155     text = open(file, 'r').read()
 156     f = StringIO()
 157     for line in lines[:-2]:
 158         f.write(line)
 159     f.writelines(lines[-2:])
 160     if f.getvalue() != text:
 161         raise RuntimeError, 'write failed'
 162     length = f.tell()
 163     print 'File length =', length
 164     f.seek(len(lines[0]))
 165     f.write(lines[1])
 166     f.seek(0)
 167     print 'First line =', `f.readline()`
 168     here = f.tell()
 169     line = f.readline()
 170     print 'Second line =', `line`
 171     f.seek(-len(line), 1)
 172     line2 = f.read(len(line))
 173     if line != line2:
 174         raise RuntimeError, 'bad result after seek back'
 175     f.seek(len(line2), 1)
 176     list = f.readlines()
 177     line = list[-1]
 178     f.seek(f.tell() - len(line))
 179     line2 = f.read()
 180     if line != line2:
 181         raise RuntimeError, 'bad result after seek back from EOF'
 182     print 'Read', len(list), 'more lines'
 183     print 'File length =', f.tell()
 184     if f.tell() != length:
 185         raise RuntimeError, 'bad length'
 186     f.close()
 187
 188 if __name__ == '__main__':
 189     _test()
 190 """