Skip to content

Instantly share code, notes, and snippets.

@cbsmith
Last active August 29, 2015 13:56
Show Gist options
  • Select an option

  • Save cbsmith/9053866 to your computer and use it in GitHub Desktop.

Select an option

Save cbsmith/9053866 to your computer and use it in GitHub Desktop.
An attempt to help this reddit poster use databases more effectively: http://www.reddit.com/r/Python/comments/1y521t/using_dictionaries_to_join_to_sql_results/
#!/usr/bin/env python
import sqlite3
from_location_group = {
'OBD1': 'Online',
'OBD2': 'Online',
'CB07': 'Retail',
'CB08': 'Retail',
'CB09': 'Retail',
'CB10': 'Retail',
'CB11': 'Retail'
}
testdata = (
('OBD1', 'order1'),
('OBD2', 'order2'),
('CB07', 'order3'),
('CB08', 'order4'),
('CB09', 'order5'),
('CB10', 'order6'),
('CB11', 'order7'),
('CB07', 'order8'),
('CB08', 'order9'),
('OBD1', 'order10'),
('OBD2', 'order11'),
('OBD1', 'order12'),
('OBD2', 'order13'),
('OBD1', 'order14'),
('OBD2', 'order15'),
('OBD3', 'order16') # intentionally not in dict to see how
# to manage left outer join style query
)
def get_rows(connection):
#probably should make this a generator, but don't want to confuse you with complexity of using a context manager with a generator
with cur as connection.cursor():
cur.execute('select orderid, location, description from order;')
return cur.fetchall():
def creates(connection, location_group_dict, testdata):
try:
with cur as connection.cursor():
cur.execute('drop table if exists order_location')
cur.execute('drop table if exists location')
cur.execute('create table location (location varchar PRIMARY KEY, description varchar);')
cur.execute('create table order_location (location varchar, orderid varchar, foreign key (location) references location.location);')
cur.execute('create view if if not exists order as (select order_location.location as location, orderid, description from order_location join location_group on (order_location.location = location.location));')
cur.executemany('insert into location_group (location, description) values (?,?)', location_group_dict.iteritems())
if testdata:
cur.executemany('insert into order_location (location, orderid) values (?,?)', testdata)
except:
connection.rollback()
else:
connection.commit()
def main():
with con as (sqlite3.connect('dicttest.db' if len(sys.argv) <= 1 else sys.argv[1])):
creates(con, from_location_group, testdata)
for row in get_rows():
print ','.join(row)
print '\n'
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment