Added first version of generator script.

generate.py

+#!/usr/bin/env python
+import re
+import MySQLdb as mysql
+from singplur import singularize, pluralize
+from itertools import combinations
+
+
+PRIMARY_KEY = 'id'
+TAB = '\t'
+MODEL_NAMESPACE = ''
+
+
+def php_value(data, indent=0):
+    """
+    Convert Python data types to PHP code. Tuples or lists are used for
+    single-line arrays, sets for multiline arrays, dictionaries for associative
+    arrays, and strings are encapsulated in single quotes.
+    """
+    if isinstance(data, str):
+        return "'%s'" % data.replace("'", "\\'")
+
+    tabs = (indent + 1) * TAB
+
+    if type(data) in (list, tuple):
+        return 'array(%s)' % ', '.join(map(php_value, data))
+    elif isinstance(data, dict):
+        if not data:
+            return 'array()'
+
+        lines = ["'%s' => %s" % (k, php_value(v, indent + 1))
+                 for k, v in data.iteritems()]
+    elif isinstance(data, set):
+        if not data:
+            return 'array()'
+
+        lines = [php_value(item, indent + 1) for item in data]
+    else:
+        return str(data)
+
+    return 'array(\n%s\n%s)' % (',\n'.join([tabs + line for line in lines]),
+                                indent * TAB)
+
+
+def php_assoc((arg, kwargs)):
+    assocs = ["'%s' => %s" % (k, php_value(v)) for k, v in kwargs.iteritems()]
+    return 'array(%s)' % ', '.join([php_value(arg)] + assocs)
+
+
+def php_static_var(name, value):
+    return 'static $%s = %s;' % (name, value)
+
+
+def php_static_array(name, values):
+    array = 'array(%s)' if len(values) < 2 \
+            else 'array(\n{0}{0}%s\n{0})'.format(TAB)
+    lines = map(php_assoc, values)
+    array_values = (',\n' + 2 * TAB).join(lines)
+    return php_static_var(name, array % array_values)
+
+
+def php_block(php_code):
+    return '<?php\n\n%s\n\n?>' % php_code
+
+
+class Model(object):
+    def __init__(self, table, options):
+        self.table = table
+        self.options = options
+
+        self.attributes = []
+        self.primary_key = []
+        self.foreign_keys = []
+
+        self.has_many = []
+        self.has_one = []
+        self.belongs_to = []
+
+    def __str__(self):
+        return '<Model "%s" table=%s>' % (self.name(), self.table)
+
+    def singular_name(self):
+        return singularize(self.table)
+
+    def classname(self):
+        return singularize(re.sub('(?:^|_)([a-z])',
+            lambda m: m.group(1).upper(), self.table))
+
+    def read_attributes(self, conn):
+        fields = read_fields(conn, self.table)
+        is_protected = lambda field: field['Key'] in ('PRI', 'MUL')
+        self.protected_fields, accessible = partition(is_protected, fields)
+        self.accessible_attr = [field['Field'] for field in accessible]
+
+    def process_keys(self, models):
+        for field in self.protected_fields:
+            name = field['Field']
+
+            if field['Key'] == 'PRI':
+                self.primary_key.append(name)
+            elif field['Key'] == 'MUL':
+                self.add_foreign_key(name, models)
+
+        # Has-many through
+        if len(self.belongs_to) > 1:
+            relations = [find_model(name, models)
+                         for name, options in self.belongs_to]
+
+            for a, b in combinations(relations, 2):
+                a.add_has_relation(b, through=self.table)
+                b.add_has_relation(a, through=self.table)
+
+    def add_has_relation(self, model, **kwargs):
+        options = self.relopts(model)
+        options.update(kwargs)
+        name = model.singular_name()
+
+        if [self.singular_name(), name] in self.options.get('has_one', []):
+            self.has_one.append((name, options))
+        else:
+            self.has_many.append((model.table, options))
+
+    def relopts(self, model, **kwargs):
+        relopts = kwargs
+
+        if self.options.get('create_select', True):
+            relopts['select'] = model.accessible_attr
+
+        return relopts
+
+    def add_foreign_key(self, name, models):
+        self.foreign_keys.append(name)
+        singular = re.sub('_' + PRIMARY_KEY + '$', '', name)
+        other_model = find_model(singular, models)
+
+        self.belongs_to.append((singular, self.relopts(other_model)))
+        other_model.add_has_relation(self)
+
+    def php_has_many(self):
+        if self.has_many:
+            return php_static_array('has_many', self.has_many)
+
+    def php_has_one(self):
+        if self.has_one:
+            return php_static_array('has_one', self.has_one)
+
+    def php_belongs_to(self):
+        if self.belongs_to:
+            return php_static_array('belongs_to', self.belongs_to)
+
+    def php_has_many_and_belongs_to(self):
+        if not self.has_many or not self.belongs_to:
+            return
+
+    def php_attr_accessible(self):
+        # All attributes except primary/foreign keys are accessible
+        if self.accessible_attr:
+            return php_static_var('attr_accessible',
+                                  php_value(self.accessible_attr))
+
+    def generate_php(self, add_php_block=False):
+        # PHP delimiters
+        php = '<?php\n\n%s\n\n?>'
+
+        # Class definition
+        classname = self.classname()
+
+        if self.options.get('namespace'):
+            classname = options['namespace'] + '\\' + classname
+
+        classdef = 'class %s extends ActiveRecord\\Model {\n%%s\n}' % classname
+
+        # Indented lines within class definition
+        lines = []
+
+        if self.options.get('create_relations', True):
+            lines += [
+                self.php_has_many(),
+                self.php_has_one(),
+                self.php_belongs_to(),
+                self.php_has_many_and_belongs_to(),
+            ]
+
+        if self.options.get('create_accessible', True):
+           lines.append(self.php_attr_accessible())
+
+        php = classdef % '\n'.join([TAB + line for line in lines if line != None])
+
+        return php_block(php) if add_php_block else php
+
+    def filename(self):
+        return self.singular_name() + '.php'
+
+    def create_path_from_dir(self, dirname):
+        return os.path.dirname(dirname + '//') + '/' + self.filename()
+
+    def save_in_file(self, path):
+        code = self.generate_php(True)
+        f = open(path, 'w')
+        f.write(code)
+        f.close()
+
+
+def find_model(singular_name, models):
+    for model in models:
+        if model.singular_name() == singular_name:
+            return model
+
+    raise ValueError('No model found for "%s".' % singular_name)
+
+
+def flatten(iterable):
+    return reduce(lambda a, b: a + b, map(list, iterable), [])
+
+
+def create_models(conn, options):
+    models = [Model(table, options) for table in read_tables(conn)]
+
+    for model in models:
+        model.read_attributes(conn)
+
+    for model in models:
+        model.process_keys(models)
+
+    return models
+
+
+def refine_model(conn, model, models):
+    for field in read_fields(conn, model.table):
+        model.add_attribute(field, models)
+
+
+def read_tables(conn):
+    cur = conn.cursor()
+    cur.execute('show tables')
+    tables = flatten(cur.fetchall())
+    cur.close()
+    return tables
+
+
+def read_fields(conn, table):
+    cur = conn.cursor(mysql.cursors.DictCursor)
+    cur.execute('show columns from %s' % table)
+    fields = cur.fetchall()
+    cur.close()
+    return fields
+
+
+def partition(callback, iterable):
+    """
+    Partition an iterable into two parts using a callback that returns a
+    boolean.
+
+    Example:
+    >>> partition(lambda x: x & 1, range(6))
+    ([1, 3, 5], [0, 2, 4])
+    """
+    a, b = [], []
+
+    for item in iterable:
+        (a if callback(item) else b).append(item)
+
+    return a, b
+
+
+if __name__ == '__main__':
+    import os
+    from argparse import ArgumentParser
+
+    parser = ArgumentParser(description='Generate PHPActiveRecord models.')
+    parser.add_argument('dbname', help='database name')
+    parser.add_argument('-H', '--host', metavar='ADDRESS', default='localhost',
+                        help='MySQL server address')
+    parser.add_argument('-u', '--user', metavar='USERNAME', default='root',
+                        help='MySQL username')
+    parser.add_argument('-p', '--password', default='mysql12#$',
+                        help='MySQL password')
+    parser.add_argument('--has-one', nargs=2, dest='opt_has_one', action='append',
+                        help='one-to-one relationships of the form '
+                        '\'payment-receipt\', where one payment has one '
+                        'receipt')
+    parser.add_argument('--namespace', default='', dest='opt_namespace',
+                        help='PHP namespace to create models classes in')
+    parser.add_argument('--create-select', action='store_true',
+                        help='whether to skip creation of the \'select\' '
+                             'option in relations', dest='opt_create_select')
+    parser.add_argument('--create-accessible', dest='opt_create_accessible',
+                        help='whether to skip creation of the '
+                             '\'$attr_accessible\' variable, containing all '
+                             'attributes except primary or foreing keys',
+                        action='store_true')
+    parser.add_argument('-a', '--create-all', action='store_true',
+                        help='create all available options and variables')
+    parser.add_argument('-d', '--dir', help='directory to save model files in')
+    args = parser.parse_args()
+    options = {}
+
+    for arg, value in args._get_kwargs():
+        if arg[:4] == 'opt_':
+            options[arg[4:]] = value
+
+    if args.create_all:
+        options['create_select'] = options['create_accessible'] = True
+
+    conn = mysql.connect(args.host, args.user, args.password, args.dbname)
+    models = create_models(conn, options)
+    conn.close()
+
+    if args.dir:
+        if not os.path.exists(args.dir):
+            os.makedirs(args.dir)
+
+        for model in models:
+            path = model.create_path_from_dir(args.dir)
+            model.save_in_file(path)
+            print 'Saved model %s in %s' % (model.classname(), path)
+    else:
+        for model in models:
+            print model.generate_php()

tests/test_generate.py

+from unittest import TestCase
+import MySQLdb as mysql
+from generate import Model, php_value, php_assoc, flatten, read_tables, read_fields
+
+
+class GenerateTest(TestCase):
+    conn = mysql.connect('localhost', 'root', 'mysql12#$', 'test')
+
+    def setUp(self):
+        self.model = Model('payments')
+
+    def test_php_value_dict(self):
+        self.assertEqual(php_value({}), 'array()')
+        self.assertEqual(php_value({'foo': 1}), "array(\n\t'foo' => 1\n)")
+        self.assertEqual(php_value({'foo': 1, 'bar': 'baz'}), """array(
+\t'foo' => 1,
+\t'bar' => 'baz'
+)""")
+        self.assertEqual(php_value({'foo': tuple(range(3))}), """array(
+\t'foo' => array(0, 1, 2)
+)""")
+        self.assertMultiLineEqual(php_value({'foo': set(range(2))}), """array(
+\t'foo' => array(
+\t\t0,
+\t\t1
+\t)
+)""")
+
+    def test_php_value_tuple(self):
+        self.assertEqual(php_value(()), 'array()')
+        self.assertEqual(php_value((1,)), 'array(1)')
+        self.assertEqual(php_value((1, 'foo')), "array(1, 'foo')")
+
+    def test_php_value_set(self):
+        self.assertEqual(php_value(set()), 'array()')
+        self.assertEqual(php_value(set([1])), 'array(\n\t1\n)')
+        self.assertEqual(php_value(set([1, 2])), 'array(\n\t1,\n\t2\n)')
+
+    def test_php_value_non_iterable(self):
+        self.assertEqual(php_value('foo'), "'foo'")
+        self.assertEqual(php_value("foo's"), "'foo\\'s'")
+        self.assertEqual(php_value(1), '1')
+
+    def test_php_assoc(self):
+        self.assertEqual(php_assoc(('users', {'through': 'payments'})),
+                         "array('users', 'through' => 'payments')")
+
+    def test_flatten(self):
+        self.assertEqual(flatten([['a'], ['b']]), ['a', 'b'])
+        self.assertEqual(flatten([['a'], ['b'], ('c', 'd')]), list('abcd'))
+
+    def test_read_tables(self):
+        self.assertEqual(read_tables(self.conn), ['payments', 'users'])
+
+    def test_read_fields(self):
+        fields = read_fields(self.conn, 'payments')
+        self.assertEqual([field['Field'] for field in fields],
+                         ['id', 'amount', 'user_id'])
+
+    def test_model_read_attributes(self):
+        self.model.read_attributes(self.conn)
+        self.assertEqual(self.model.accessible_attr, ['amount'])