I'm trying to implement a REST api to returns records from a SQL database in CSV format. I would like to use the import csv
to do the conversion and I would like to use a generator so that I can stream the results.
This is what I have tried.
def get_data(query) -> Generator[str, None, None]:
with pyodbc.connect(connStr) as conn:
with conn.cursor() as cursor:
cursor.execute(query)
while 1:
row = cursor.fetchone()
if not row: break
data = io.StringIO()
csv.writer(data).writerow(row)
yield data.getvalue()
This works, but I don't like it. It seems too verbose and it creates so many temporary StringIO and writer objects!! Please tell me a better way.
CodePudding user response:
I suppose I could reuse the StringIO using a seek...
def get_data(query) -> Generator[str, None, None]:
data = io.StringIO()
writer = csv.writer(data)
with pyodbc.connect(connStr) as conn:
with conn.cursor() as cursor:
cursor.execute(query)
while 1:
row = cursor.fetchone()
if not row: break
data.seek(0)
writer.writerow(row)
yield data.getvalue()
CodePudding user response:
If the goal is just return/yield a resultset record values (gathered as tuple) in csv format you don't need overcomplicating with io.StringIO()
and csv.writer
:
def get_data(query) -> Generator[str, None, None]:
with pyodbc.connect(connStr) as conn:
with conn.cursor() as cursor:
cursor.execute(query)
while True:
row = cursor.fetchone()
if not row: break
yield ', '.join(map(str, (row)))