Skip to content

Feature csvfile merge #22

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MarvKler wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Feature csvfile merge #22

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b717980
refactor package / added keyword to merge CSV files
Mar 24, 2025
12d437e
d
Mar 24, 2025
a085104
revert
Mar 24, 2025
7ccd424
fix correct version info
Mar 25, 2025
8302a79
fix some linting mentions
Mar 25, 2025
15ccfb9
split helper from keyword
Mar 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,13 @@
# RF / VSCode
output.xml
report.html
log.html
launch.json
01*.csv
02*.csv
03*.csv
04*.csv

# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
Expand Down
301 changes: 1 addition & 300 deletions CSVLibrary/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,300 +1 @@
import csv
import sys

from robot.api import logger
from robot.utils.dotdict import DotDict

if sys.version_info.major >= 3:
from io import StringIO as IO
else:
from io import BytesIO as IO

__version__ = '0.0.5'

class DotDictReader(csv.DictReader):
def __next__(self):
return DotDict(super().__next__())


class CSVLibrary(object):

ROBOT_LIBRARY_VERSION = __version__
ROBOT_LIBRARY_SCOPE = 'GLOBAL'

@staticmethod
def _reader(to_read, csv_reader=csv.reader, line_numbers=None, **kwargs):
reader = csv_reader(to_read, **kwargs)
try:
for line_number, row in enumerate(reader):
if line_numbers is None:
yield row
elif isinstance(line_numbers, list) and line_number in line_numbers:
yield row
line_numbers.remove(line_number)
if len(line_numbers) == 0:
break
except csv.Error as e:
logger.error('line %d: %s' % (reader.line_num, e))

def _read_csv(self, csv_handler, csv_reader=csv.reader, line_numbers=None, **kwargs):
if line_numbers is not None and isinstance(line_numbers, list):
line_numbers = list(map(int, line_numbers))

return [row for row in self._reader(csv_handler, csv_reader, line_numbers, **kwargs)]

def _open_csv_file_for_read(self, filename, csv_reader=csv.reader, line_numbers=None, **kwargs):
with open(filename, 'r') as csv_handler:
return self._read_csv(csv_handler, csv_reader, line_numbers, **kwargs)

@staticmethod
def _write_csv(csv_handler, data, csv_writer=csv.writer, **kwargs):
if 'fieldnames' not in kwargs.keys() and isinstance(data[0], dict):
kwargs['fieldnames'] = data[0].keys()

writer = csv_writer(csv_handler, **kwargs)
try:
if isinstance(writer, csv.DictWriter) and csv_handler.tell() == 0:
writer.writeheader()

if not isinstance(data[0], (list, tuple, dict, set)):
data = [data]
writer.writerows(data)
except csv.Error as e:
logger.error('%s' % e)

def _open_csv_file_for_write(self, filename, data, csv_writer=csv.writer, **kwargs):
open_args = {'mode': 'ab'}
if sys.version_info >= (3, 2):
open_args['mode'] = 'a'
open_args['newline'] = ''

with open(filename, **open_args) as csv_handler:
self._write_csv(csv_handler, data, csv_writer, **kwargs)

@staticmethod
def empty_csv_file(filename):
"""This keyword will empty the CSV file.

- ``filename``: name of csv file
"""
with open(filename, "w") as csv_handler:
csv_handler.truncate()

def read_csv_file_to_list(self, filename, delimiter=',', **kwargs):
"""Read CSV file and return its content as a Python list of tuples.

- ``filename``: name of csv file
- ``delimiter``: Default: `,`
- ``line_numbers``: List of linenumbers to read. Default None
- ``quoting`` (int):
_0_: QUOTE_MINIMAL
_1_: QUOTE_ALL
_2_: QUOTE_NONNUMERIC
_3_: QUOTE_NONE
"""
csv_list = self._open_csv_file_for_read(
filename,
csv_reader=csv.reader,
delimiter=str(delimiter),
**kwargs
)
return csv_list

def read_csv_string_to_list(self, csv_string, delimiter=',', **kwargs):
"""Read CSV string and return its content as a Python list of tuples.

- ``csv_string``: name of csv file
- ``delimiter``: Default: `,`
- ``line_numbers``: List of linenumbers to read. Default None
- ``quoting`` (int):
_0_: QUOTE_MINIMAL
_1_: QUOTE_ALL
_2_: QUOTE_NONNUMERIC
_3_: QUOTE_NONE
"""
if sys.version_info.major < 3:
csv_string = csv_string.encode("utf-8")

with IO(csv_string) as csv_handler:
csv_list = self._read_csv(
csv_handler,
csv_reader=csv.reader,
delimiter=str(delimiter),
**kwargs
)
return csv_list

def read_csv_file_to_associative(self, filename, delimiter=',', fieldnames=None, **kwargs):
"""Read CSV file and return its content as a Python list of dictionaries.

- ``filename``: name of csv file
- ``delimiter``: Default: `,`
- ``fieldnames``: list of column names
- ``line_numbers``: List of linenumbers to read. Default None
- ``quoting`` (int):
_0_: QUOTE_MINIMAL
_1_: QUOTE_ALL
_2_: QUOTE_NONNUMERIC
_3_: QUOTE_NONE
"""
kwargs['fieldnames'] = fieldnames
csv_dict = self._open_csv_file_for_read(
filename,
csv_reader=DotDictReader,
delimiter=str(delimiter),
**kwargs
)
return csv_dict

def read_csv_string_to_associative(self, csv_string, delimiter=',', fieldnames=None, **kwargs):
"""Read CSV from string and return its content as a Python list of dictionaries.

- ``csv_string``: csv formatted string
- ``delimiter``: Default: `,`
- ``fieldnames``: list of column names
- ``line_numbers``: List of linenumbers to read. Default None
- ``quoting`` (int):
_0_: QUOTE_MINIMAL
_1_: QUOTE_ALL
_2_: QUOTE_NONNUMERIC
_3_: QUOTE_NONE
"""
if sys.version_info.major < 3:
csv_string = csv_string.encode("utf-8")

with IO(csv_string) as csv_handler:
csv_dict = self._read_csv(
csv_handler,
csv_reader=DotDictReader,
delimiter=str(delimiter),
fieldnames=fieldnames,
**kwargs
)
return csv_dict

def append_to_csv_file(self, filename, data, **kwargs):
"""This keyword will append data to a new or existing CSV file.

- ``filename``: name of csv file
- ``data``: iterable(e.g. list or tuple) data.
- ``quoting`` (int):
_0_: QUOTE_MINIMAL
_1_: QUOTE_ALL
_2_: QUOTE_NONNUMERIC
_3_: QUOTE_NONE
"""

header = self._open_csv_file_for_read(
filename,
csv_reader=csv.reader,
line_numbers=[0],
**kwargs
)

if isinstance(data, dict):
data = [data]

fieldnames = None
if isinstance(data[0], dict):
fieldnames = data[0].keys()

if isinstance(data[0], dict) and len(header) > 0:
fieldnames = header[0]

if fieldnames is not None:
kwargs['fieldnames'] = fieldnames
kwargs['csv_writer'] = csv.DictWriter

self._open_csv_file_for_write(
filename,
data=data,
**kwargs
)

def append_to_csv_string(self, csv_string, data, **kwargs):
"""This keyword will append data to a new or existing CSV string.

- ``csv_string``: csv formatted string
- ``data``: iterable(e.g. list or tuple) data.
- ``quoting`` (int):
_0_: QUOTE_MINIMAL
_1_: QUOTE_ALL
_2_: QUOTE_NONNUMERIC
_3_: QUOTE_NONE
"""
if isinstance(data, dict):
data = [data]

if 'lineterminator' not in kwargs.keys():
kwargs['lineterminator'] = '\n'

if sys.version_info.major < 3:
csv_string = csv_string.encode("utf-8")

with IO(csv_string) as csv_handler:
header = self._read_csv(
csv_handler,
csv_reader=csv.reader,
**kwargs
)

fieldnames = None
if isinstance(data[0], dict):
fieldnames = data[0].keys()

if isinstance(data[0], dict) and len(header) > 0:
fieldnames = header[0]

if fieldnames is not None:
kwargs['fieldnames'] = fieldnames
kwargs['csv_writer'] = csv.DictWriter

self._write_csv(csv_handler, data, **kwargs)
return csv_handler.getvalue()

def csv_file_from_associative(self, filename, data, fieldnames=None, delimiter=',', **kwargs):
"""This keyword will create new file

- ``filename``: name of csv file
- ``data``: iterable(e.g. list or tuple) data.
- ``fieldnames``: list of column names
- ``delimiter``: Default: `,`
- ``quoting`` (int):
_0_: QUOTE_MINIMAL
_1_: QUOTE_ALL
_2_: QUOTE_NONNUMERIC
_3_: QUOTE_NONE
"""
kwargs['fieldnames'] = fieldnames or data[0].keys()

self._open_csv_file_for_write(
filename,
data=data,
csv_writer=csv.DictWriter,
delimiter=str(delimiter),
**kwargs
)

def csv_string_from_associative(self, data, fieldnames=None, delimiter=',', **kwargs):
"""This keyword will return csv string

- ``data``: iterable(e.g. list or tuple) data.
- ``fieldnames``: list of column names
- ``delimiter``: Default: `,`
- ``quoting`` (int):
_0_: QUOTE_MINIMAL
_1_: QUOTE_ALL
_2_: QUOTE_NONNUMERIC
_3_: QUOTE_NONE
"""
if isinstance(data, dict):
data = [data]

kwargs['fieldnames'] = fieldnames or data[0].keys()
if 'lineterminator' not in kwargs.keys():
kwargs['lineterminator'] = '\n'
kwargs['delimiter'] = delimiter

with IO() as csv_handler:
self._write_csv(csv_handler, data, csv_writer=csv.DictWriter, **kwargs)
return csv_handler.getvalue()
from .keywords import Keywords as CSVLibrary
Loading