Skip to content

Commit

Permalink
Update sql releated documentation and tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@chfw
chfw committed May 13, 2015
1 parent 5a10e0a commit 549a5f1
Show file tree
Hide file tree
Showing 4 changed files with 212 additions and 24 deletions.
134 changes: 125 additions & 9 deletions doc/source/sqlalchemy.rst
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,11 +53,9 @@ Now let's verify the data:
>>> from pyexcel_io import from_query_sets
>>> query_sets=mysession.query(Pyexcel).all()
>>> results = from_query_sets(data[0], query_sets)
>>> import pprint
>>> pprint.pprint(results)
[['birth', 'id', 'name', 'weight'],
['2014-11-11', 0, u'Adam', 11.25],
['2014-11-12', 1, u'Smith', 12.25]]
>>> import json
>>> json.dumps(results)
'[["birth", "id", "name", "weight"], ["2014-11-11", 0, "Adam", 11.25], ["2014-11-12", 1, "Smith", 12.25]]'


Read data from a database table
Expand All @@ -68,9 +66,127 @@ Let's use previous data for reading and see if we could get them via

>>> from pyexcel_io import get_data
>>> data = get_data(DB_SQL, session=mysession, tables=[Pyexcel])
>>> pprint.pprint(data)
[['birth', 'id', 'name', 'weight'],
['2014-11-11', 0, u'Adam', 11.25],
['2014-11-12', 1, u'Smith', 12.25]]
>>> json.dumps(results)
'[["birth", "id", "name", "weight"], ["2014-11-11", 0, "Adam", 11.25], ["2014-11-12", 1, "Smith", 12.25]]'


Write data to multiple tables
--------------------------------------------------------------------------------

Before we start, let's clear off previous table:

>>> Base.metadata.drop_all(engine)

Now suppose we have these more complex tables:

>>> from sqlalchemy.orm import relationship, backref
>>> class Post(Base):
... __tablename__ = 'post'
... id = Column(Integer, primary_key=True)
... title = Column(String(80))
... body = Column(String(100))
... pub_date = Column(DateTime)
...
... category_id = Column(Integer, ForeignKey('category.id'))
... category = relationship('Category',
... backref=backref('posts', lazy='dynamic'))
...
... def __init__(self, title, body, category, pub_date=None):
... self.title = title
... self.body = body
... if pub_date is None:
... pub_date = datetime.utcnow()
... self.pub_date = pub_date
... self.category = category
...
... def __repr__(self):
... return '<Post %r>' % self.title
...
>>> class Category(Base):
... __tablename__ = 'category'
... id = Column(Integer, primary_key=True)
... name = Column(String(50))
...
... def __init__(self, name):
... self.name = name
...
... def __repr__(self):
... return '<Category %r>' % self.name
... def __str__(self):
... return self.__repr__()

Let's clear the database and create previous table in the database:

>>> Base.metadata.create_all(engine)

Suppose we have these data:

>>> data = {
... "Category":[
... ["id", "name"],
... [1, "News"],
... [2, "Sports"]
... ],
... "Post":[
... ["id", "title", "body", "pub_date", "category"],
... [1, "Title A", "formal", datetime.datetime(2015,1,20,23,28,29), "News"],
... [2, "Title B", "informal", datetime.datetime(2015,1,20,23,28,30), "Sports"]
... ]
... }

Both table has gotten initialization functions:

>>> def category_init_func(row):
... c = Category(row['name'])
... c.id = row['id']
... return c

and particularly **Post** has a foreign key to **Category**, so we need to
query **Category** out and assign it to **Post** instance

>>> def post_init_func(row):
... c = mysession.query(Category).filter_by(name=row['category']).first()
... p = Post(row['title'], row['body'], c, row['pub_date'])
... return p

Here's the code to update both:

>>> tables = {
... "Category": [Category, data['Category'][0], None, category_init_func],
... "Post": [Post, data['Post'][0], None, post_init_func]
... }
>>> to_store = {
... "Category": data['Category'][1:],
... "Post": data['Post'][1:]
... }
>>> save_data(DB_SQL, to_store, session=mysession, tables=tables)

Let's verify what do we have in the database:

>>> query_sets = mysession.query(Category).all()
>>> results = from_query_sets(data['Category'][0], query_sets)
>>> import json
>>> json.dumps(results)
'[["id", "name"], [1, "News"], [2, "Sports"]]'
>>> query_sets = mysession.query(Post).all()
>>> results = from_query_sets(["id", "title", "body", "pub_date"], query_sets)
>>> json.dumps(results)
'[["id", "title", "body", "pub_date"], [1, "Title A", "formal", "2015-01-20T23:28:29"], [2, "Title B", "informal", "2015-01-20T23:28:30"]]'


Read data from multiple tables
----------------------------------------------------------------------------------

Let's use previous data for reading and see if we could get them via
:meth:`~pyexcel_io.get_data` :

>>> data = get_data(DB_SQL, session=mysession, tables=[Category, Post])
>>> json.dumps(data)
'{"category": [["id", "name"], [1, "News"], [2, "Sports"]], "post": [["body", "category_id", "id", "pub_date", "title"], ["formal", 1, 1, "2015-01-20T23:28:29", "Title A"], ["informal", 2, 2, "2015-01-20T23:28:30", "Title B"]]}'

.. testcode::
:hide:

>>> mysession.close()
>>> import os
>>> os.unlink('tmp.db')
3 changes: 2 additions & 1 deletion test.bat
View file
Original file line number Diff line number Diff line change
@@ -1 +1,2 @@
nosetests --with-cov --cov pyexcel_io --cov tests --with-doctest --doctest-extension=.rst doc/source
nosetests --with-cov --cov pyexcel_io --cov tests --with-doctest --doctest-extension=.rst doc/source tests pyexcel_io
del tmp.db
1 change: 1 addition & 0 deletions test.sh
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
#/bin/bash

nosetests --rednose --with-cov --cov pyexcel_io --cov tests --with-doctest --doctest-extension=.rst doc/source tests pyexcel_io
rm tmp.db
98 changes: 84 additions & 14 deletions tests/test_sql_book.py
View file
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column , Integer, String, Float, Date
from sqlalchemy import Column , Integer, String, Float, Date, DateTime, ForeignKey
from sqlalchemy.orm import sessionmaker
import datetime
from pyexcel_io import DB_SQL, get_data, save_data, from_query_sets
from sqlalchemy.orm import relationship, backref

engine=create_engine("sqlite:///tmp.db")
Base=declarative_base()
Expand All @@ -15,21 +16,44 @@ class Pyexcel(Base):
weight=Column(Float)
birth=Column(Date)

class Signature(Base):
__tablename__="signature"
X=Column(Integer, primary_key=True)
Y=Column(Integer)
Z=Column(Integer)

class Signature2(Base):
__tablename__="signature2"
A=Column(Integer, primary_key=True)
B=Column(Integer)
C=Column(Integer)
class Post(Base):
__tablename__ = 'post'
id = Column(Integer, primary_key=True)
title = Column(String(80))
body = Column(String(100))
pub_date = Column(DateTime)

category_id = Column(Integer, ForeignKey('category.id'))
category = relationship('Category',
backref=backref('posts', lazy='dynamic'))

def __init__(self, title, body, category, pub_date=None):
self.title = title
self.body = body
if pub_date is None:
pub_date = datetime.utcnow()
self.pub_date = pub_date
self.category = category

def __repr__(self):
return '<Post %r>' % self.title

class Category(Base):
__tablename__ = 'category'
id = Column(Integer, primary_key=True)
name = Column(String(50))

def __init__(self, name):
self.name = name

def __repr__(self):
return '<Category %r>' % self.name
def __str__(self):
return self.__repr__()

Session=sessionmaker(bind=engine)

class TestSQL:
class TestSingleRead:
def setUp(self):
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
Expand Down Expand Up @@ -59,7 +83,7 @@ def test_sql(self):
mysession.close()


class TestWritingSQL:
class TestSingleWrite:
def setUp(self):
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
Expand All @@ -84,3 +108,49 @@ def test_one_table(self):
query_sets=mysession.query(Pyexcel).all()
results = from_query_sets(self.data[0], query_sets)
assert results == self.results



class TestMultipleRead:
def setUp(self):
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
self.session = Session()
data = {
"Category":[
["id", "name"],
[1, "News"],
[2, "Sports"]
],
"Post":[
["id", "title", "body", "pub_date", "category"],
[1, "Title A", "formal", datetime.datetime(2015,1,20,23,28,29), "News"],
[2, "Title B", "informal", datetime.datetime(2015,1,20,23,28,30), "Sports"]
]
}
def category_init_func(row):
c = Category(row['name'])
c.id = row['id']
return c

def post_init_func(row):
c = self.session.query(Category).filter_by(name=row['category']).first()
p = Post(row['title'], row['body'], c, row['pub_date'])
return p
tables = {
"Category": [Category, data['Category'][0], None, category_init_func],
"Post": [Post, data['Post'][0], None, post_init_func]
}
to_store = {
"Category": data['Category'][1:],
"Post": data['Post'][1:]
}
save_data(DB_SQL, to_store, session=self.session, tables=tables)

def test_read(self):
data = get_data(DB_SQL, session=self.session, tables=[Category, Post])
import json
assert json.dumps(data) == '{"category": [["id", "name"], [1, "News"], [2, "Sports"]], "post": [["body", "category_id", "id", "pub_date", "title"], ["formal", 1, 1, "2015-01-20T23:28:29", "Title A"], ["informal", 2, 2, "2015-01-20T23:28:30", "Title B"]]}'

def tearDown(self):
self.session.close()

0 comments on commit 549a5f1

Please sign in to comment.