Source code for sqlobject.util.csvimport

Import from a CSV file or directory of files.

CSV files should have a header line that lists columns.  Headers can
also be appended with ``:type`` to indicate the type of the field.
``escaped`` is the default, though it can be overridden by the importer.
Supported types:

    A python expression, run through ``eval()``.  This can be a
    security risk, pass in ``allow_python=False`` if you don't want to
    allow it.




    A string with backslash escapes (note that you don't put quotation
    marks around the value)

    A base64-encoded string

    ISO date, like YYYY-MM-DD; this can also be ``NOW+days`` or

    ISO date/time like YYYY-MM-DDTHH:MM:SS (either T or a space can be
    used to separate the time, and seconds are optional).  This can
    also be ``NOW+seconds`` or ``NOW-seconds``

    Converts true/false/yes/no/on/off/1/0 to boolean value

    This will be resolved to the ID of the object named in this column
    (None if the column is empty).  @@: Since there's no ordering,
    there's no way to promise the object already exists.

You can also get back references to the objects if you have a special
``[name]`` column.

Any column named ``[comment]`` or with no name will be ignored.

In any column you can put ``[default]`` to exclude the value and use
whatever default the class wants.  ``[null]`` will use NULL.

Lines that begin with ``[comment]`` are ignored.

import csv
from datetime import datetime, date, timedelta
import os
import time
import types

__all__ = ['load_csv_from_directory',

DEFAULT_TYPE = 'escaped'

[docs]def create_data(data, class_getter, keyorder=None): """ Create the ``data``, which is the return value from ``load_csv()``. Classes will be resolved with the callable ``class_getter``; or if ``class_getter`` is a module then the class names will be attributes of that. Returns a dictionary of ``{object_name: object(s)}``, using the names from the ``[name]`` columns (if there are any). If a name is used multiple times, you get a list of objects, not a single object. If ``keyorder`` is given, then the keys will be retrieved in that order. It can be a list/tuple of names, or a sorting function. If not given and ``class_getter`` is a module and has a ``soClasses`` function, then that will be used for the order. """ objects = {} classnames = data.keys() if (not keyorder and isinstance(class_getter, types.ModuleType) and hasattr(class_getter, 'soClasses')): keyorder = [c.__name__ for c in class_getter.soClasses] if not keyorder: classnames.sort() elif isinstance(keyorder, (list, tuple)): all = classnames classnames = [name for name in keyorder if name in classnames] for name in all: if name not in classnames: classnames.append(name) else: classnames.sort(keyorder) for classname in classnames: items = data[classname] if not items: continue if isinstance(class_getter, types.ModuleType): soClass = getattr(class_getter, classname) else: soClass = class_getter(classname) for item in items: for key, value in item.items(): if isinstance(value, Reference): resolved = objects.get( if not resolved: raise ValueError( "Object reference to %r does not have target" % elif (isinstance(resolved, list) and len(resolved) > 1): raise ValueError( "Object reference to %r is ambiguous (got %r)" % (, resolved)) item[key] = if '[name]' in item: name = item.pop('[name]').strip() else: name = None inst = soClass(**item) if name: if name in objects: if isinstance(objects[name], list): objects[name].append(inst) else: objects[name] = [objects[name], inst] else: objects[name] = inst return objects
[docs]def load_csv_from_directory(directory, allow_python=True, default_type=DEFAULT_TYPE, allow_multiple_classes=True): """ Load the data from all the files in a directory. Filenames indicate the class, with ``general.csv`` for data not associated with a class. Return data just like ``load_csv`` does. This might cause problems on case-insensitive filesystems. """ results = {} for filename in os.listdir(directory): base, ext = os.path.splitext(filename) if ext.lower() != '.csv': continue f = open(os.path.join(directory, filename), 'rb') csvreader = csv.reader(f) data = load_csv(csvreader, allow_python=allow_python, default_type=default_type, default_class=base, allow_multiple_classes=allow_multiple_classes) f.close() for classname, items in data.items(): results.setdefault(classname, []).extend(items) return results
[docs]def load_csv(csvreader, allow_python=True, default_type=DEFAULT_TYPE, default_class=None, allow_multiple_classes=True): """ Loads the CSV file, returning a list of dictionaries with types coerced. """ current_class = default_class current_headers = None results = {} for row in csvreader: if not [cell for cell in row if cell.strip()]: # empty row continue if row and row[0].strip() == 'CLASS:': if not allow_multiple_classes: raise ValueError( "CLASS: line in CSV file, but multiple classes " "are not allowed in this file (line: %r)" % row) if not row[1:]: raise ValueError( "CLASS: in line in CSV file, with no class name " "in next column (line: %r)" % row) current_class = row[1] current_headers = None continue if not current_class: raise ValueError( "No CLASS: line given, and there is no default class " "for this file (line: %r)" % row) if current_headers is None: current_headers = _parse_headers(row, default_type, allow_python=allow_python) continue if row[0] == '[comment]': continue # Pad row with empty strings: row += [''] * (len(current_headers) - len(row)) row_converted = {} for value, (name, coercer, args) in zip(row, current_headers): if name is None: # Comment continue if value == '[default]': continue if value == '[null]': row_converted[name] = None continue args = (value,) + args row_converted[name] = coercer(*args) results.setdefault(current_class, []).append(row_converted) return results
def _parse_headers(header_row, default_type, allow_python=True): headers = [] for name in header_row: original_name = name if ':' in name: name, type = name.split(':', 1) else: type = default_type if type == 'python' and not allow_python: raise ValueError( ":python header given when python headers are not allowed " "(with header %r)" % original_name) name = name.strip() if name == '[comment]' or not name: headers.append((None, None, None)) continue type = type.strip().lower() if '(' in type: type, arg = type.split('(', 1) if not arg.endswith(')'): raise ValueError( "Arguments (in ()'s) do not end with ): %r" % original_name) args = (arg[:-1],) else: args = () if name == '[name]': type = 'str' coercer, args = get_coercer(type) headers.append((name, coercer, args)) return headers _coercers = {} def get_coercer(type): if type not in _coercers: raise ValueError( "Coercion type %r not known (I know: %s)" % (type, ', '.join(_coercers.keys()))) return _coercers[type] def register_coercer(type, coercer, *args): _coercers[type] = (coercer, args) def identity(v): return v register_coercer('str', identity) register_coercer('string', identity) def decode_string(v, encoding): return v.decode(encoding) register_coercer('escaped', decode_string, 'string_escape') register_coercer('strescaped', decode_string, 'string_escape') register_coercer('base64', decode_string, 'base64') register_coercer('int', int) register_coercer('float', float) def parse_python(v): return eval(v, {}, {}) register_coercer('python', parse_python) def parse_date(v): v = v.strip() if not v: return None if v.startswith('NOW-') or v.startswith('NOW+'): days = int(v[3:]) now = return now + timedelta(days) else: parsed = time.strptime(v, '%Y-%m-%d') return date.fromtimestamp(time.mktime(parsed)) register_coercer('date', parse_date) def parse_datetime(v): v = v.strip() if not v: return None if v.startswith('NOW-') or v.startswith('NOW+'): seconds = int(v[3:]) now = return now + timedelta(0, seconds) else: fmts = ['%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M', '%Y-%m-%d %H:%M'] for fmt in fmts[:-1]: try: parsed = time.strptime(v, fmt) break except ValueError: pass else: parsed = time.strptime(v, fmts[-1]) return datetime.fromtimestamp(time.mktime(parsed)) register_coercer('datetime', parse_datetime) class Reference(object): def __init__(self, name): = name def parse_ref(v): if not v.strip(): return None else: return Reference(v) register_coercer('ref', parse_ref) def parse_bool(v): v = v.strip().lower() if v in ('y', 'yes', 't', 'true', 'on', '1'): return True elif v in ('n', 'no', 'f', 'false', 'off', '0'): return False raise ValueError( "Value is not boolean-like: %r" % v) register_coercer('bool', parse_bool) register_coercer('boolean', parse_bool)