1
# Copyright (C) 2005 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
# \subsection{\emph{rio} - simple text metaformat}
19
# \emph{r} stands for `restricted', `reproducible', or `rfc822-like'.
21
# The stored data consists of a series of \emph{stanzas}, each of which contains
22
# \emph{fields} identified by an ascii name, with Unicode or string contents.
23
# The field tag is constrained to alphanumeric characters.
24
# There may be more than one field in a stanza with the same name.
26
# The format itself does not deal with character encoding issues, though
27
# the result will normally be written in Unicode.
29
# The format is intended to be simple enough that there is exactly one character
30
# stream representation of an object and vice versa, and that this relation
31
# will continue to hold for future versions of bzr.
35
from bzrlib.iterablefile import IterableFile
37
# XXX: some redundancy is allowing to write stanzas in isolation as well as
38
# through a writer object.
40
class RioWriter(object):
41
def __init__(self, to_file):
43
self._to_file = to_file
45
def write_stanza(self, stanza):
48
stanza.write(self._to_file)
52
class RioReader(object):
53
"""Read stanzas from a file as a sequence
55
to_file can be anything that can be enumerated as a sequence of
56
lines (with newlines.)
58
def __init__(self, from_file):
59
self._from_file = from_file
63
s = read_stanza(self._from_file)
70
def rio_file(stanzas, header=None):
71
"""Produce a rio IterableFile from an iterable of stanzas"""
73
if header is not None:
77
if first_stanza is not True:
79
for line in s.to_lines():
82
return IterableFile(str_iter())
85
def read_stanzas(from_file):
87
s = read_stanza(from_file)
94
"""One stanza for rio.
96
Each stanza contains a set of named fields.
98
Names must be non-empty ascii alphanumeric plus _. Names can be repeated
99
within a stanza. Names are case-sensitive. The ordering of fields is
102
Each field value must be either an int or a string.
105
__slots__ = ['items']
107
def __init__(self, **kwargs):
108
"""Construct a new Stanza.
110
The keyword arguments, if any, are added in sorted order to the stanza.
114
for tag, value in sorted(kwargs.items()):
117
def add(self, tag, value):
118
"""Append a name and value to the stanza."""
119
assert valid_tag(tag), \
120
("invalid tag %r" % tag)
121
if isinstance(value, str):
122
value = unicode(value)
123
elif isinstance(value, unicode):
125
## elif isinstance(value, (int, long)):
126
## value = str(value) # XXX: python2.4 without L-suffix
128
raise TypeError("invalid type for rio value: %r of type %s"
129
% (value, type(value)))
130
self.items.append((tag, value))
132
def __contains__(self, find_tag):
133
"""True if there is any field in this stanza with the given tag."""
134
for tag, value in self.items:
140
"""Return number of pairs in the stanza."""
141
return len(self.items)
143
def __eq__(self, other):
144
if not isinstance(other, Stanza):
146
return self.items == other.items
148
def __ne__(self, other):
149
return not self.__eq__(other)
152
return "Stanza(%r)" % self.items
154
def iter_pairs(self):
155
"""Return iterator of tag, value pairs."""
156
return iter(self.items)
159
"""Generate sequence of lines for external version of this file.
161
The lines are always utf-8 encoded strings.
164
# max() complains if sequence is empty
167
for tag, value in self.items:
168
assert isinstance(tag, str), type(tag)
169
assert isinstance(value, unicode)
171
result.append(tag + ': \n')
173
# don't want splitlines behaviour on empty lines
174
val_lines = value.split('\n')
175
result.append(tag + ': ' + val_lines[0].encode('utf-8') + '\n')
176
for line in val_lines[1:]:
177
result.append('\t' + line.encode('utf-8') + '\n')
179
result.append(tag + ': ' + value.encode('utf-8') + '\n')
183
"""Return stanza as a single string"""
184
return ''.join(self.to_lines())
186
def to_unicode(self):
187
"""Return stanza as a single Unicode string.
189
This is most useful when adding a Stanza to a parent Stanza
195
for tag, value in self.items:
197
result.append(tag + ': \n')
199
# don't want splitlines behaviour on empty lines
200
val_lines = value.split('\n')
201
result.append(tag + ': ' + val_lines[0] + '\n')
202
for line in val_lines[1:]:
203
result.append('\t' + line + '\n')
205
result.append(tag + ': ' + value + '\n')
206
return u''.join(result)
208
def write(self, to_file):
209
"""Write stanza to a file"""
210
to_file.writelines(self.to_lines())
213
"""Return the value for a field wih given tag.
215
If there is more than one value, only the first is returned. If the
216
tag is not present, KeyError is raised.
218
for t, v in self.items:
226
def get_all(self, tag):
228
for t, v in self.items:
234
"""Return a dict containing the unique values of the stanza.
237
for tag, value in self.items:
242
_tag_re = re.compile(r'^[-a-zA-Z0-9_]+$')
244
return bool(_tag_re.match(tag))
247
def read_stanza(line_iter):
248
"""Return new Stanza read from list of lines or a file
250
Returns one Stanza that was read, or returns None at end of file. If a
251
blank line follows the stanza, it is consumed. It's not an error for
252
there to be no blank at end of file. If there is a blank file at the
253
start of the input this is really an empty stanza and that is returned.
255
Only the stanza lines and the trailing blank (if any) are consumed
258
The raw lines must be in utf-8 encoding.
260
unicode_iter = (line.decode('utf-8') for line in line_iter)
261
return read_stanza_unicode(unicode_iter)
264
def read_stanza_unicode(unicode_iter):
265
"""Read a Stanza from a list of lines or a file.
267
The lines should already be in unicode form. This returns a single
268
stanza that was read. If there is a blank line at the end of the Stanza,
269
it is consumed. It is not an error for there to be no blank line at
270
the end of the iterable. If there is a blank line at the beginning,
271
this is treated as an empty Stanza and None is returned.
273
Only the stanza lines and the trailing blank (if any) are consumed
274
from the unicode_iter
276
:param unicode_iter: A iterable, yeilding Unicode strings. See read_stanza
277
if you have a utf-8 encoded string.
278
:return: A Stanza object if there are any lines in the file.
285
# TODO: jam 20060922 This code should raise real errors rather than
286
# using 'assert' to process user input, or raising ValueError
287
# rather than a more specific error.
289
for line in unicode_iter:
290
if line is None or line == '':
293
break # end of stanza
294
assert line.endswith('\n')
296
if line[0] == '\t': # continues previous value
298
raise ValueError('invalid continuation line %r' % real_l)
299
accum_value += '\n' + line[1:-1]
300
else: # new tag:value line
302
stanza.add(tag, accum_value)
304
colon_index = line.index(': ')
306
raise ValueError('tag/value separator not found in line %r'
308
tag = str(line[:colon_index])
309
assert valid_tag(tag), \
310
"invalid rio tag %r" % tag
311
accum_value = line[colon_index+2:-1]
313
if tag is not None: # add last tag-value
314
stanza.add(tag, accum_value)
316
else: # didn't see any content
added
removed
bzrlib/rio.py
