forked from SvenskaSpel/locust-plugins
-
Notifications
You must be signed in to change notification settings - Fork 0
/
csvreader.py
43 lines (35 loc) · 1.19 KB
/
csvreader.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import csv
from typing import Iterator, Dict
class CSVReader(Iterator):
"Read test data from csv file using an iterator"
def __init__(self, file, **kwargs):
try:
file = open(file)
except TypeError:
pass # "file" was already a pre-opened file-like object
self.file = file
self.reader = csv.reader(file, **kwargs)
def __next__(self):
try:
return next(self.reader)
except StopIteration:
# reuse file on EOF
self.file.seek(0, 0)
return next(self.reader)
class CSVDictReader(Iterator[Dict]):
"Read test data from csv file using an iterator, returns rows as dicts"
def __init__(self, file, **kwargs):
try:
file = open(file)
except TypeError:
pass # "file" was already a pre-opened file-like object
self.file = file
self.reader = csv.DictReader(file, **kwargs)
def __next__(self):
try:
return next(self.reader)
except StopIteration:
# reuse file on EOF
self.file.seek(0, 0)
next(self.reader) # skip header line
return next(self.reader)