Add export CSV data route
issue9090 review261011002
This commit is contained in:
Cédric Krier
@@ -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
|
||||
|
||||
@@ -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``.
|
||||
|
||||
@@ -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 = '''<!DOCTYPE html>
|
||||
</form>
|
||||
</body>
|
||||
</html>'''
|
||||
|
||||
|
||||
@app.route('/<database_name>/data/<model>', 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
|
||||
|
||||
@@ -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):
|
||||
'''
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
@@ -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')
|
||||
|
||||
Reference in New Issue
Block a user