Convert SQL query to CSV file in Python
This function takes an SQLAlchemy query object as it’s input (SQLAlchemy is a popular Python SQL toolkit and Object Relational Mapper). It’s strength lies in not needing to hard-code column names, making it scaleable and suitable to “set-and-forget”.
While developing enterprise software, every developer has surely have been faced with a client asking “how do I get the data out of the system?”, and after digging a bit deeper, it becomes clear they want the ability to save a snapshot of data as an Excel spreadsheet. While it may be a good idea in the longer term to sit down with the client to figure out how the application might be extended to automate these additional steps, here is a handy python script that will “get the data” from a SQL database interfaced with SQLAlchemy and output a CSV file.
sql_query_to_csv.py
def sql_query_to_csv(query_output, columns_to_exclude=""):
""" Converts output from a SQLAlchemy query to a .csv string.
Parameters:
query_output (list of <class 'SQLAlchemy.Model'>): output from an SQLAlchemy query.
columns_to_exclude (list of str): names of columns to exclude from .csv output.
Returns:
csv (str): query_output represented in .csv format.
Example usage:
users = db.Users.query.filter_by(user_id=123)
csv = sql_query_to_csv(users, ["id", "age", "address"]
"""
rows = query_output
columns_to_exclude = set(columns_to_exclude)
#create list of column names
column_names = [i for i in rows[0].__dict__]
for column_name in columns_to_exclude:
column_names.pop(column_names.index(column_name))
#add column titles to csv
column_names.sort()
csv = ", ".join(column_names) + "\n"
#add rows of data to csv
for row in rows:
for column_name in column_names:
if column_name not in columns_to_exclude:
data = str(row.__dict__[column_name])
#Escape (") symbol by preceeding with another (")
data.replace('"','""')
#Enclose each datum in double quotes so commas within are not treated as separators
csv += '"' + data + '"' + ","
csv += "\n"
return csv
Example implementation
from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy import Column, String, Integer, DateTime
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.declarative import declarative_base
from sql_query_to_csv import sql_query_to_csv
Base = declarative_base()
# define model
class User(Base):
__tablename__ = 'User'
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
fullname = Column(String, nullable=False)
birth = Column(DateTime)
engine = create_engine('sqlite:///db.sqlite')
Base.metadata.create_all(bind=engine)
users = [
User(name='JH',
fullname='Jimi Hendrix',
birth=datetime(1942,11,27)),
User(name='RJ',
fullname='Robert Johnson',
birth=datetime(1943,12,8)),
User(name='JM',
fullname='Jim Morrison',
birth=datetime(1911,5,8))]
# create session
Session = sessionmaker()
Session.configure(bind=engine)
session = Session()
# add data
session.add_all(users)
session.commit()
# query database
query = session.query(User).all()
# output all users to csv
csv_1 = sql_query_to_csv(query)
print(csv_1)
# output all users to csv, excluding some columns
csv_2 = sql_query_to_csv(query, ["birth", "id", "_sa_instance_state"])
print(csv_2)
Example output
_sa_instance_state, birth, fullname, id, name
"<sqlalchemy.orm.state.InstanceState object at 0x00000230B14AB2C8>","1942-11-27 00:00:00","Jimi Hendrix","1","JH",
"<sqlalchemy.orm.state.InstanceState object at 0x00000230B07985C8>","1943-12-08 00:00:00","Robert Johnson","2","RJ",
"<sqlalchemy.orm.state.InstanceState object at 0x00000230B14D8C48>","1911-05-08 00:00:00","Jim Morrison","3","JM",
fullname, name
"Jimi Hendrix","JH",
"Robert Johnson","RJ",
"Jim Morrison","JM",