From 5626018cd60d287a9aca1f34f3e5ebdb25bd79c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9dric=20Krier?= Date: Sun, 12 Apr 2020 23:10:30 +0200 Subject: [PATCH] Add export CSV data route issue9090 review261011002 --- CHANGELOG | 2 + doc/ref/models/models.rst | 5 + trytond/ir/routes.py | 92 ++++++++++++++ trytond/model/modelstorage.py | 7 ++ trytond/tests/test_exportdata.py | 17 +++ trytond/tests/test_routes.py | 201 +++++++++++++++++++++++++++++++ trytond/tests/test_tryton.py | 6 +- 7 files changed, 328 insertions(+), 2 deletions(-) create mode 100644 trytond/tests/test_routes.py diff --git a/CHANGELOG b/CHANGELOG index 79b775e61b..dd649638d1 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,5 @@ +* Add export_data_domain to ModelStorage +* Add route to export CSV data * Add test on target of relation fields * Add width and height attributes on calendar view * Add default database name configuration diff --git a/doc/ref/models/models.rst b/doc/ref/models/models.rst index cb1f98b36c..d03ab49c2c 100644 --- a/doc/ref/models/models.rst +++ b/doc/ref/models/models.rst @@ -326,6 +326,11 @@ Class methods: Descriptor on fields are available by appending ``.`` and the name of the method on the field that returns the descriptor. +.. classmethod:: ModelStorage.export_data_domain(domain, fields_names[, offset[, limit[, order]]]) + + Call :meth:`search` and :meth`export_data` together. + Useful for the client to reduce the number of calls and the data transfered. + .. classmethod:: ModelStorage.import_data(fields_names, data) Create or update records for all values in ``data``. diff --git a/trytond/ir/routes.py b/trytond/ir/routes.py index 5d4d6c2406..bc7e89f35c 100644 --- a/trytond/ir/routes.py +++ b/trytond/ir/routes.py @@ -1,5 +1,10 @@ # This file is part of Tryton. The COPYRIGHT file at the top level of # this repository contains the full copyright notices and license terms. +import csv +import datetime as dt +import json +import io +from numbers import Number try: from http import HTTPStatus except ImportError: @@ -12,7 +17,9 @@ from werkzeug.wrappers import Response from trytond.i18n import gettext from trytond.config import config from trytond.wsgi import app +from trytond.protocols.jsonrpc import JSONDecoder from trytond.protocols.wrappers import with_pool, with_transaction +from trytond.tools import slugify from trytond.transaction import Transaction SOURCE = config.get( @@ -161,3 +168,88 @@ TEMPLATE = ''' ''' + + +@app.route('//data/', methods={'GET'}) +@app.auth_required +@with_pool +@with_transaction(user='request', context=dict(_check_access=True)) +def data(request, pool, model): + User = pool.get('res.user') + Lang = pool.get('ir.lang') + try: + Model = pool.get(model) + except KeyError: + abort(HTTPStatus.NOT_FOUND) + transaction = Transaction() + context = User(transaction.user).get_preferences(context_only=True) + language = request.args.get('l') + if language: + context['language'] = language + try: + domain = json.loads( + request.args.get('d', '[]'), object_hook=JSONDecoder()) + except json.JSONDecodeError: + abort(HTTPStatus.BAD_REQUEST) + limit = None + offset = 0 + if 's' in request.args: + try: + limit = int(request.args.get('s')) + if 'p' in request.args: + offset = int(request.args.get('p')) * limit + except ValueError: + abort(HTTPStatus.BAD_REQUEST) + if 'o' in request.args: + order = [(o.split(',', 1) + [''])[:2] + for o in request.args.getlist('o')] + else: + order = None + fields_names = request.args.getlist('f') + encoding = request.args.get('enc', 'UTF-8') + delimiter = request.args.get('dl', ',') + quotechar = request.args.get('qc', '"') + try: + header = bool(int(request.args.get('h', True))) + locale_format = bool(int(request.args.get('loc', False))) + except ValueError: + abort(HTTPStatus.BAD_REQUEST) + + with transaction.set_context(**context): + lang = Lang.get(transaction.language) + + def format_(row): + for i, value in enumerate(row): + if locale_format: + if isinstance(value, Number): + value = lang.format('%.12g', value) + elif isinstance(value, (dt.date, dt.datetime)): + value = lang.strftime(value) + elif isinstance(value, bool): + value = int(value) + row[i] = value + return row + + try: + if domain and isinstance(domain[0], (int, float)): + rows = Model.export_data(domain, fields_names) + else: + rows = Model.export_data_domain( + domain, fields_names, + limit=limit, offset=offset, order=order) + except (ValueError, KeyError): + abort(HTTPStatus.BAD_REQUEST) + data = io.StringIO(newline='') + writer = csv.writer(data, delimiter=delimiter, quotechar=quotechar) + if header: + writer.writerow(fields_names) + for row in rows: + writer.writerow(format_(row)) + data = data.getvalue().encode(encoding) + filename = slugify(Model.__names__()['model']) + '.csv' + filename = filename.encode('latin-1', 'ignore') + response = Response(data, mimetype='text/csv; charset=' + encoding) + response.headers.add( + 'Content-Disposition', 'attachment', filename=filename) + response.headers.add('Content-Length', len(data)) + return response diff --git a/trytond/model/modelstorage.py b/trytond/model/modelstorage.py index af68bd6f48..9b19a3a663 100644 --- a/trytond/model/modelstorage.py +++ b/trytond/model/modelstorage.py @@ -124,6 +124,7 @@ class ModelStorage(Model): 'search_count': RPC(), 'search_read': RPC(), 'resources': RPC(instantiate=0, unique=False), + 'export_data_domain': RPC(), 'export_data': RPC(instantiate=0, unique=False), 'import_data': RPC(readonly=False), }) @@ -676,6 +677,12 @@ class ModelStorage(Model): data += cls.__export_row(record, fields_names) return data + @classmethod + def export_data_domain( + cls, domain, fields_names, offset=0, limit=None, order=None): + records = cls.search(domain, limit=limit, offset=offset, order=order) + return cls.export_data(records, fields_names) + @classmethod def import_data(cls, fields_names, data): ''' diff --git a/trytond/tests/test_exportdata.py b/trytond/tests/test_exportdata.py index b18200ce52..fc45cc6f49 100644 --- a/trytond/tests/test_exportdata.py +++ b/trytond/tests/test_exportdata.py @@ -360,6 +360,23 @@ class ExportDataTestCase(unittest.TestCase): ExportData.export_data([export1], ['reference.translated']), [["Target"]]) + @with_transaction() + def test_domain(self): + "Test export data with domain" + pool = Pool() + ExportData = pool.get('text.export_data') + + ExportData.create([{ + 'boolean': True, + }, { + 'boolean': False, + }]) + + self.assertEqual( + ExportData.export_data_domain( + [('boolean', '=', True)], ['boolean']), + [[True]]) + def suite(): return unittest.TestLoader().loadTestsFromTestCase(ExportDataTestCase) diff --git a/trytond/tests/test_routes.py b/trytond/tests/test_routes.py new file mode 100644 index 0000000000..8e0b726694 --- /dev/null +++ b/trytond/tests/test_routes.py @@ -0,0 +1,201 @@ +# This file is part of Tryton. The COPYRIGHT file at the top level of this +# repository contains the full copyright notices and license terms. + +import base64 +import json +import unittest + +from werkzeug.test import Client +from werkzeug.wrappers import BaseResponse + +from trytond.pool import Pool +from trytond.tests.test_tryton import activate_module, DB_NAME, drop_db +from trytond.transaction import Transaction +from trytond.wsgi import app + + +class RoutesTestCase(unittest.TestCase): + + @classmethod + def setUpClass(cls): + drop_db() + activate_module(['ir', 'res'], 'fr') + pool = Pool(DB_NAME) + with Transaction().start(DB_NAME, 0): + User = pool.get('res.user') + admin, = User.search([('login', '=', 'admin')]) + admin.password = 'password' + admin.save() + + @classmethod + def tearDownClass(cls): + super().tearDownClass() + drop_db() + + @property + def auth_headers(self): + return { + 'Authorization': b'Basic ' + base64.b64encode(b'admin:password'), + } + + def data_url(self, model): + return '/%(database)s/data/%(model)s' % { + 'database': DB_NAME, + 'model': model, + } + + def test_data_no_field(self): + "Test GET data without field" + c = Client(app, BaseResponse) + + response = c.get(self.data_url('res.user'), headers=self.auth_headers) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.data, b'\r\n\r\n') + + def test_data_one_field(self): + "Test GET data with one field" + c = Client(app, BaseResponse) + + response = c.get( + self.data_url('res.user'), headers=self.auth_headers, + query_string=[('f', 'name')]) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.data, b'name\r\nAdministrator\r\n') + + def test_data_multiple_fields(self): + "Test GET data with multiple fields" + c = Client(app, BaseResponse) + + response = c.get( + self.data_url('res.user'), headers=self.auth_headers, + query_string=[('f', 'name'), ('f', 'login')]) + + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.data, b'name,login\r\nAdministrator,admin\r\n') + + def test_data_language(self): + "Test GET data with language" + c = Client(app, BaseResponse) + + response = c.get( + self.data_url('ir.lang'), headers=self.auth_headers, + query_string=[ + ('f', 'name'), + ('l', 'fr'), + ('d', json.dumps([('code', '=', 'fr')])), + ]) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.data, 'name\r\nFrançais\r\n'.encode('utf-8')) + + def test_data_size(self): + "Test GET data with size limit" + c = Client(app, BaseResponse) + + response = c.get( + self.data_url('ir.lang'), headers=self.auth_headers, + query_string=[ + ('f', 'name'), + ('s', 5), + ]) + + self.assertEqual(response.status_code, 200) + self.assertEqual(len(response.data.splitlines()), 5 + 1) + + def test_data_page(self): + "Test GET data with page" + c = Client(app, BaseResponse) + + response0 = c.get( + self.data_url('ir.lang'), headers=self.auth_headers, + query_string=[ + ('f', 'name'), + ('s', 5), + ('p', 0) + ]) + response1 = c.get( + self.data_url('ir.lang'), headers=self.auth_headers, + query_string=[ + ('f', 'name'), + ('s', 5), + ('p', 1) + ]) + + self.assertEqual(response0.status_code, 200) + self.assertEqual(response1.status_code, 200) + self.assertNotEqual(response0.data, response1.data) + + def test_data_encoding(self): + "Test GET data with encoding" + c = Client(app, BaseResponse) + + response = c.get( + self.data_url('ir.lang'), headers=self.auth_headers, + query_string=[ + ('f', 'name'), + ('l', 'fr'), + ('d', json.dumps([('code', '=', 'fr')])), + ('enc', 'latin1'), + ]) + + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.data, 'name\r\nFrançais\r\n'.encode('latin1')) + + def test_data_delimiter(self): + "Test GET data with delimiter" + c = Client(app, BaseResponse) + + response = c.get( + self.data_url('res.user'), headers=self.auth_headers, + query_string=[('f', 'name'), ('f', 'login'), ('dl', '|')]) + + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.data, b'name|login\r\nAdministrator|admin\r\n') + + def test_data_quotechar(self): + "Test GET data with quotechar" + c = Client(app, BaseResponse) + + response = c.get( + self.data_url('res.user'), headers=self.auth_headers, + query_string=[ + ('f', 'name'), ('f', 'login'), ('dl', 'n'), ('qc', '*')]) + + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.data, b'*name*n*login*\r\n*Administrator*n*admin*\r\n') + + def test_data_no_header(self): + "Test GET data without header" + c = Client(app, BaseResponse) + + response = c.get( + self.data_url('res.user'), headers=self.auth_headers, + query_string=[('f', 'name'), ('h', 0)]) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.data, b'Administrator\r\n') + + def test_data_locale_format(self): + "Test GET data in locale format" + c = Client(app, BaseResponse) + + response_std = c.get( + self.data_url('res.user'), headers=self.auth_headers, + query_string=[('f', 'create_date')]) + response_locale = c.get( + self.data_url('res.user'), headers=self.auth_headers, + query_string=[('f', 'create_date'), ('loc', 1)]) + + self.assertEqual(response_std.status_code, 200) + self.assertEqual(response_locale.status_code, 200) + self.assertNotEqual(response_std.data, response_locale.data) + + +def suite(): + return unittest.TestLoader().loadTestsFromTestCase(RoutesTestCase) diff --git a/trytond/tests/test_tryton.py b/trytond/tests/test_tryton.py index ec47e9dac2..e38f99fe25 100644 --- a/trytond/tests/test_tryton.py +++ b/trytond/tests/test_tryton.py @@ -45,16 +45,18 @@ DB_CACHE = os.environ.get('DB_CACHE') Pool.test = True -def activate_module(modules): +def activate_module(modules, lang='en'): ''' Activate modules for the tested database ''' if isinstance(modules, str): modules = [modules] name = '-'.join(modules) + if lang != 'en': + name += '--' + lang if not db_exist(DB_NAME) and restore_db_cache(name): return - create_db() + create_db(lang=lang) with Transaction().start(DB_NAME, 1, close=True) as transaction: pool = Pool() Module = pool.get('ir.module')