[pycopia] r741 committed - Fixes to work with the latest sqlalchemy. Now requires sqlalchemy 0.9.

[pyc...@googlecode.com]

Revision: 741
Author: keith.dart
Date: Mon Mar 3 19:58:36 2014 UTC
Log: Fixes to work with the latest sqlalchemy. Now requires sqlalchemy
0.9.


http://code.google.com/p/pycopia/source/detail?r=741

Modified:
/trunk/storage/init_db.py
/trunk/storage/pycopia/db/models.py
/trunk/storage/pycopia/db/tui/widgets.py
/trunk/storage/pycopia/db/types.py
/trunk/storage/setup.py

=======================================
--- /trunk/storage/init_db.py Fri Sep 28 23:40:30 2012 UTC
+++ /trunk/storage/init_db.py Mon Mar 3 19:58:36 2014 UTC
@@ -20,8 +20,10 @@
"""

import sys
+import os

from pycopia import aid
+from pycopia import urlparse
from pycopia.db import models
from pycopia.db import config

@@ -498,13 +500,13 @@
def create_db(url):
url = urlparse.UniversalResourceLocator(url, True)
scheme = url.scheme
- if scheme == "postgres":
+ if scheme == "postgresql":
cmd = 'sudo su postgres -c "createuser --host %s --createdb
--no-superuser --no-createrole %s"' % (url.host, url.user)
os.system(cmd)
cmd = 'sudo su postgres -c "createdb --host %s --owner %s
--encoding utf-8 %s"' % (url.host, url.user, url.path[1:])
os.system(cmd)
else:
- raise NotImplementedError
+ raise NotImplementedError("unhandled scheme: {}".format(scheme))


def do_config(session):
@@ -562,6 +564,7 @@
from pycopia import basicconfig
cf = basicconfig.get_config("database.conf")
url = cf["DATABASE_URL"]
+ create_db(url)
dbsession = models.get_session(url)
try:
do_schedules(dbsession)
=======================================
--- /trunk/storage/pycopia/db/models.py Fri Aug 2 07:49:54 2013 UTC
+++ /trunk/storage/pycopia/db/models.py Mon Mar 3 19:58:36 2014 UTC
@@ -31,7 +31,7 @@
from datetime import timedelta
from hashlib import sha1

-from sqlalchemy import create_engine, and_, or_, not_, func, exists
+from sqlalchemy import create_engine, inspect, and_, or_, not_, func,
exists
from sqlalchemy.orm import (sessionmaker, mapper, relationship,
class_mapper,
backref, synonym, _mapper_registry, validates)
from sqlalchemy.orm.properties import ColumnProperty, RelationshipProperty
@@ -1626,7 +1626,7 @@

def get_primary_key(class_):
"""Return the primary key column."""
- return class_mapper(class_).primary_key
+ return inspect(class_).primary_key


def get_primary_key_value(dbrow):
@@ -1639,8 +1639,7 @@

def get_primary_key_name(class_):
"""Return name or names of primary key column. Return None if not
defined."""
- mapper = class_mapper(class_)
- pk = mapper.primary_key
+ pk = inspect(class_).primary_key
pk_l = len(pk)
if pk_l == 0:
return None
@@ -1650,55 +1649,12 @@
return tuple(p.name for p in pk)


-def get_choices(session, modelclass, colname, order_by=None):
- """Get possible choices for a field.
-
- Returns a list of tuples, (id/value, name/label) of available choices.
- """
-
- # first check for column type with get_choices method.
- mapper = class_mapper(modelclass)
- try:
- return mapper.columns[colname].type.get_choices()
- except (AttributeError, KeyError):
- pass
- mycol = getattr(modelclass, colname)
- try:
- relmodel = mycol.property.mapper.class_
- except AttributeError:
- return []
-
- # no choices in type, but has a related model table...
- mymeta = get_column_metadata(modelclass, colname)
- if mymeta.uselist:
- if mymeta.m2m:
- q = session.query(relmodel)
- else:
- # only those that are currently unassigned
- rs = mycol.property.remote_side[0]
- q = session.query(relmodel).filter(rs == None)
- else:
- q = session.query(relmodel)
-
- # add optional order_by, default to ordering by first ROW_DISPLAY
column.
- try:
- order_by = order_by or relmodel.ROW_DISPLAY[0]
- except (AttributeError, IndexError):
- pass
- if order_by:
- q = q.order_by(getattr(relmodel, order_by))
- # Return the list of (id, stringrep) tuples.
- # This structure is usable by the XHTML form generator...
- return [(relrow.id, str(relrow)) for relrow in q]
-
-
# structure returned by get_metadata function.
MetaDataTuple = collections.namedtuple("MetaDataTuple",
"coltype, colname, default, m2m, nullable, uselist, collection")

-
def get_metadata_iterator(class_):
- for prop in class_mapper(class_).iterate_properties:
+ for prop in inspect(class_).iterate_properties:
name = prop.key
if name.startswith("_") or name == "id" or name.endswith("_id"):
continue
@@ -1707,15 +1663,13 @@
continue
yield md

-
def get_column_metadata(class_, colname):
- prop = class_mapper(class_).get_property(colname)
+ prop = inspect(class_).get_property(colname)
md = _get_column_metadata(prop)
if md is None:
raise ValueError("Not a column name: %r." % (colname,))
return md

-
def _get_column_metadata(prop):
name = prop.key
m2m = False
@@ -1737,7 +1691,7 @@
elif proptype is RelationshipProperty:
coltype = RelationshipProperty.__name__
m2m = prop.secondary is not None
- nullable = prop.local_side[0].nullable
+ nullable = prop.local_remote_pairs[0][0].nullable
uselist = prop.uselist
if prop.collection_class is not None:
collection = type(prop.collection_class()).__name__
@@ -1747,6 +1701,46 @@
return None
return MetaDataTuple(coltype, str(name), default, m2m, nullable,
uselist, collection)

+
+def get_choices(session, modelclass, colname, order_by=None):
+ """Get possible choices for a field.
+
+ Returns a list of tuples, (id/value, name/label) of available choices.
+ """
+ # first check for column type with get_choices method.
+ mapper = inspect(modelclass)
+ try:
+ return mapper.columns[colname].type.get_choices()
+ except (AttributeError, KeyError):
+ pass
+ mycol = getattr(modelclass, colname)
+ try:
+ relmodel = mycol.property.mapper.class_
+ except AttributeError:
+ return []
+ # no choices in type, but has a related model table...
+ mymeta = get_column_metadata(modelclass, colname)
+ if mymeta.uselist:
+ if mymeta.m2m:
+ q = session.query(relmodel)
+ else:
+ # only those that are currently unassigned
+ rs = mycol.property.local_remote_pairs[0][1]
+ q = session.query(relmodel).filter(rs == None)
+ else:
+ q = session.query(relmodel)
+
+ # add optional order_by, default to ordering by first ROW_DISPLAY
column.
+ try:
+ order_by = order_by or relmodel.ROW_DISPLAY[0]
+ except (AttributeError, IndexError):
+ pass
+ if order_by:
+ q = q.order_by(getattr(relmodel, order_by))
+ # Return the list of (id, stringrep) tuples.
+ # This structure is usable by the XHTML form generator...
+ return [(relrow.id, str(relrow)) for relrow in q]
+

def get_metadata(class_):
"""Returns a list of MetaDataTuple structures.
@@ -1770,19 +1764,23 @@
from pycopia import autodebug
if sys.flags.interactive:
from pycopia import interactive
- #prop = class_mapper(Equipment).get_property("interfaces")
- #print prop
- #print get_metadata(Equipment)
- #print get_column_metadata(Equipment, "interfaces")
- #print get_column_metadata(Network, "interfaces")
+
+ print (get_metadata(Equipment))
+
+ print(inspect(Equipment).get_property("name"))
+ assert get_primary_key_name(Equipment) == "id"
+ print(get_primary_key_name(Session))

- #sess = get_session()
+ #print (get_column_metadata(Equipment, "interfaces"))
+ #print (get_column_metadata(Network, "interfaces"))

- #choices = get_choices(sess, Equipment, "interfaces", order_by=None)
- #print choices
- #choices = get_choices(sess, TestCase, "priority", order_by=None)
- #print choices
+ sess = get_session()

+ #choices = XXXget_choices(sess, Equipment, "interfaces", order_by=None)
+ #print (choices)
+ choices = get_choices(sess, TestCase, "priority", order_by=None)
+ print (choices)
+ print (get_choices(sess, Equipment, "interfaces", order_by=None))
# assumes your host name is in the db for testing.
#eq = sess.query(Equipment).filter(Equipment.name.like(os.uname()[1]
+ "%")).one()
#print "eq = ", eq
@@ -1795,7 +1793,7 @@
# print eq.capabilities

# for res in TestResult.get_latest_results(sess):
-# print res
+# print (res)

# print "\nlatest run:"
# user = User.get_by_username(sess, "keith")
@@ -1822,12 +1820,11 @@
#
# for tr in tc.get_data(sess):
# print(tr)
- with DatabaseContext() as sess:
- for intf in Interface.select_unattached(sess):
- #print(intf)
- print(intf)
+# with DatabaseContext() as sess:
+# for intf in Interface.select_unattached(sess):
+# print(intf)
#
print(type(TestSuite.get_by_implementation(sess, "testcases.unittests.WWW.mobileget.MobileSendSuite")))
# print (TestSuite.get_suites(sess))
#print(get_primary_key(Session))
- #sess.close()
+ sess.close()

=======================================
--- /trunk/storage/pycopia/db/tui/widgets.py Thu Jan 10 00:16:24 2013 UTC
+++ /trunk/storage/pycopia/db/tui/widgets.py Mon Mar 3 19:58:36 2014 UTC
@@ -2429,7 +2429,7 @@
q = session.query(relmodel)
else:
# only those that are currently unassigned
- rs = mycol.property.remote_side[0]
+ rs = mycol.property.local_remote_pairs[0][1]
q = session.query(relmodel).filter(rs == None)
else:
q = session.query(relmodel)
@@ -2455,7 +2455,6 @@
if isinstance(value, list):
if not value:
continue
- #t =
session.query(relmodel).filter(relmodel.id.in_(value)).all()
t = value
if metadata.collection == "MappedCollection":
setattr(dbrow, metadata.colname, dict((o.name, o) for
o in t))
=======================================
--- /trunk/storage/pycopia/db/types.py Thu Jan 10 00:16:24 2013 UTC
+++ /trunk/storage/pycopia/db/types.py Mon Mar 3 19:58:36 2014 UTC
@@ -56,6 +56,7 @@
DATE, BYTEA, BOOLEAN, INTERVAL, ARRAY, ENUM)

from sqlalchemy import types
+from sqlalchemy.ext.mutable import Mutable


class ValidationError(AssertionError):
@@ -64,7 +65,7 @@

# custom column types.

-class Cidr(types.MutableType, types.TypeDecorator):
+class Cidr(Mutable, types.TypeDecorator):
"""Cidr reprsents networks without host part."""

impl = CIDR
@@ -84,7 +85,7 @@
return None
return IPv4(value)

-class Inet(types.MutableType, types.TypeDecorator):
+class Inet(Mutable, types.TypeDecorator):
"""An IPv4 address type. Columns with this type take and receive IPv4
objects from the database.
"""
@@ -254,8 +255,8 @@
return cls.enumerations.find(int(value))


-OBJECTTYPES = Enums("module", "TestSuite", "Test", "TestRunner", "unknown")
-OBJ_MODULE, OBJ_TESTSUITE, OBJ_TEST, OBJ_TESTRUNNER, OBJ_UNKNOWN =
OBJECTTYPES
+OBJECTTYPES =
Enums("UseCase", "TestSuite", "Test", "TestRunner", "unknown")
+OBJ_USECASE, OBJ_TESTSUITE, OBJ_TEST, OBJ_TESTRUNNER, OBJ_UNKNOWN =
OBJECTTYPES

class TestObjectType(types.TypeDecorator):
"""Possible test runner objects that produce results."""
=======================================
--- /trunk/storage/setup.py Wed Oct 24 16:38:51 2012 UTC
+++ /trunk/storage/setup.py Mon Mar 3 19:58:36 2014 UTC
@@ -44,7 +44,7 @@
# install_requires = [
# 'pycopia-core>=1.0.dev-r138,==dev',
# 'pycopia-CLI>=1.0.dev-r138,==dev',
-# 'sqlalchemy>=0.6.0',
+# 'sqlalchemy>=0.9.0',
# 'pycrypto>=2.0',
# 'urwid>=1.0',
# #'psycopg>=2.0',