Skip to content

Commit

Permalink
fix broken test case
Browse files Browse the repository at this point in the history
  • Loading branch information
@chfw
chfw committed May 12, 2015
1 parent 85b6444 commit 5a10e0a
Show file tree
Hide file tree
Showing 3 changed files with 88 additions and 81 deletions.
82 changes: 1 addition & 81 deletions doc/source/sqlalchemy.rst
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ Suppose we have a pure sql database connection via sqlalchemy:

>>> from sqlalchemy import create_engine
>>> from sqlalchemy.ext.declarative import declarative_base
>>> from sqlalchemy import Column , Integer, String, Float, Date, ForeignKey, backref, DateTime
>>> from sqlalchemy import Column , Integer, String, Float, Date, ForeignKey, DateTime
>>> from sqlalchemy.orm import sessionmaker
>>> engine=create_engine("sqlite:///tmp.db")
>>> Base=declarative_base()
Expand Down Expand Up @@ -74,83 +74,3 @@ Let's use previous data for reading and see if we could get them via
['2014-11-12', 1, u'Smith', 12.25]]


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

Suppose we have these more complex tables:

>>> 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

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

>>> Base.metadata.drop_all(engine)
>>> 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(2015,1,20,23,28,29), "News"],
... [2, "Title B", "informal", 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]
... }
>>> save_data(DB_SQL, data, session=mysession, tables=tables)
1 change: 1 addition & 0 deletions tests/requirements.txt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,3 +3,4 @@ rednose
nose-cov
python-coveralls
coverage
SQLAlchemy
86 changes: 86 additions & 0 deletions tests/test_sql_book.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,86 @@
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column , Integer, String, Float, Date
from sqlalchemy.orm import sessionmaker
import datetime
from pyexcel_io import DB_SQL, get_data, save_data, from_query_sets

engine=create_engine("sqlite:///tmp.db")
Base=declarative_base()

class Pyexcel(Base):
__tablename__='pyexcel'
id=Column(Integer, primary_key=True)
name=Column(String)
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)

Session=sessionmaker(bind=engine)

class TestSQL:
def setUp(self):
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
p1 = Pyexcel(id=0,
name="Adam",
weight=11.25,
birth=datetime.date(2014, 11, 11))
self.session = Session()
self.session.add(p1)
p1 = Pyexcel(id=1,
name="Smith",
weight=12.25,
birth=datetime.date(2014, 11, 12))
self.session.add(p1)
self.session.commit()
self.session.close()

def test_sql(self):
mysession=Session()
data = get_data(DB_SQL, session=mysession, tables=[Pyexcel])
content = [
['birth', 'id', 'name', 'weight'],
['2014-11-11', 0, 'Adam', 11.25],
['2014-11-12', 1, 'Smith', 12.25]
]
assert data == content
mysession.close()


class TestWritingSQL:
def setUp(self):
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
self.data = [
['birth', 'id', 'name', 'weight'],
[datetime.date(2014, 11, 11), 0, 'Adam', 11.25],
[datetime.date(2014, 11, 12), 1, 'Smith', 12.25]
]
self.results = [
['birth', 'id', 'name', 'weight'],
['2014-11-11', 0, 'Adam', 11.25],
['2014-11-12', 1, 'Smith', 12.25]
]

def test_one_table(self):
mysession = Session()
save_data(DB_SQL,
self.data[1:],
session=mysession,
tables={ 'csv': [Pyexcel,self.data[0], None, None]}
)
query_sets=mysession.query(Pyexcel).all()
results = from_query_sets(self.data[0], query_sets)
assert results == self.results

0 comments on commit 5a10e0a

Please sign in to comment.