KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use export_api method in localstack

Best Python code snippet using localstack_python

export.py

Source:export.py

1from flask import Blueprint, jsonify, request, send_file2from flask_security import login_required, current_user3from app.models import db4from app.models.observation import Observation5from app.models.station import Station6from app.models.variable import Variable7from app.models.export import Export, ExportVariable, ExportStation8from jsonschema import validate, ValidationError9from datetime import datetime, timedelta10from zipfile import ZipFile, ZIP_DEFLATED11from io import BytesIO12import pandas as pd13export_api = Blueprint('export_api', __name__)14export_schema = {15    "type": "object",16    "properties": {17        "description": {"type": "string"},18        "startDate": {"type": "string"},19        "endDate": {"type": "string"},20        "timezone": {"type": "string"},21        "stations": {22            "type": "array",23            "items": {24                "type": "string"25            }26        },27        "variables": {28            "type": "array",29            "items": {30                "type": "string"31            }32        }33    },34    "minProperties": 6,35    "additionalProperties": False36}37@export_api.route("/export", methods=['POST'])38@login_required39def create():40    content = request.get_json(silent=True)41    validate(instance=content, schema=export_schema)42    startDate = datetime.strptime(content['startDate'], "%Y-%m-%dT%H:%M:%S.%fZ")43    endDate = datetime.strptime(content['endDate'], "%Y-%m-%dT%H:%M:%S.%fZ")44    export = Export(45        user_id=current_user.id,46        aggregation="1H",47        startDate=startDate,48        endDate=endDate,49        description=content['description'],50        timezone=content['timezone'],51        variables=[ExportVariable(variable_id=int(x)) for x in content['variables']],52        stations=[ExportStation(station_id=x) for x in content['stations']]53    )54    db.add(export)55    db.commit()56    return jsonify({ "status": "success" })57@export_api.route("/export", methods=['GET'])58@login_required59def list():60    exports = db.query(Export).filter(Export.user_id == current_user.id).order_by(61        Export.created_at.desc()).all()62    return jsonify(63        {64            "status":"success",65            "exports": [66                {67                    "_id": export.id,68                    "aggregation": export.aggregation,69                    "createdAt": export.created_at,70                    "description": export.description,71                    "startDate": export.startDate,72                    "endDate": export.endDate,73                    "stations": [ s.station_id for s in export.stations ],74                    "variables": [ v.variable_id for v in export.variables ],75                    "status": "completed",76                    "user": export.user_id77                } for export in exports78            ]79         })80@export_api.route("/download/export/<id>", methods=['GET'])81@login_required82def download(id):83    # TODO: filter for current user id.84    export = Export.query.get(id)85    if not export:86        raise ValueError('Invalid export with identifier %s' % id)87    stations = Station.query.filter(Station.stationId.in_([ s.station_id for s in export.stations])).all()88    variables = Variable.query.filter(Variable.elementId.in_([ v.variable_id for v in export.variables])).all()89    timestamp_column = 'Timestamp {}'.format(export.timezone)90    # Start ZIP archive in memory.91    in_memory = BytesIO()92    zf = ZipFile(in_memory, mode="w", compression=ZIP_DEFLATED)93    for station in stations:94        dataframes = []95        for variable in variables:96            observations = Observation.query.with_entities(Observation.obsDatetime, Observation.obsValue).filter(Observation.recordedFrom == station.stationId)\97                .filter(Observation.obsDatetime >= export.startDate).filter(Observation.obsDatetime <= export.endDate)\98                .filter(Observation.describedBy == variable.elementId).filter(Observation.obsValue != -999).order_by(Observation.obsDatetime.asc()).all()99            if len(observations):100                df = pd.DataFrame.from_records(observations, index=timestamp_column, columns=[timestamp_column,'{} ({})'.format(variable.abbreviation, variable.units)])101                if export.timezone != 'UTC':102                    df = df.tz_localize('UTC').tz_convert('Africa/Nairobi')103                dataframes.append(df)104        if len(dataframes):105            zf.writestr('{}_{}.csv'.format(station.stationId, "".join(c for c in station.stationName if c.isalnum()).rstrip()),106                        pd.concat(dataframes, axis=1, join="inner").to_csv(na_rep='', date_format='%Y-%m-%d %H:%M'))107    # Add metadata file to zip archive.108    zf.writestr('metadata.csv', getMetadataCSV(stations))109    # Close the zip file and retrieve it's contents from memory.110    zf.close()111    in_memory.seek(0)112    return send_file(in_memory, attachment_filename="Export_{}.zip".format(export.id), as_attachment=True)113@export_api.errorhandler(ValidationError)114@export_api.errorhandler(ValueError)115def handle(e):116    return jsonify({"status": "error", "error": "Invalid input for export", "message": str(e)}), 400117def getMetadataCSV(stations):118    stationsMetadata = []119    for station in stations:120        stationsMetadata.append([121            station.stationId,122            station.stationName,123            station.latitude,124            station.longitude,125            station.elevation126        ])127    if len(stationsMetadata) > 0:128        metaDf = pd.DataFrame.from_records(stationsMetadata,129                                           columns=["station id", "name", "latitude", "longitude", "elevation (m)"])...

webui.py

Source:webui.py

1import logging2from deluge import component, common3from deluge.plugins.pluginbase import WebPluginBase4from deluge.ui.client import client5from deluge.ui.web.json_api import export as export_api6from twisted.internet.defer import Deferred7from .common import get_resource8LOGGER = logging.getLogger(__name__)9class WebUI(WebPluginBase):10    scripts = [get_resource('webapi.js')]11    def enable(self):12        """Triggers when plugin is enabled."""13        LOGGER.info('Enabling WebAPI plugin WEB ...')14    def disable(self):15        """Triggers when plugin is disabled."""16        LOGGER.info('Disabling WebAPI plugin WEB ...')17    @export_api18    def get_torrents(self, ids=None, params=None):19        """Returns information about all or a definite torrent.20        Returned information can be filtered by supplying wanted parameter names.21        """22        if params is None:23            filter_fields = ['name', 'comment', 'hash', 'save_path']24        else:25            if isinstance(params, list):26                filter_fields = params27            else:28                filter_fields = params.replace(" ","").split(",")29        proxy = component.get('SessionProxy')30        filter_dict = {}31        if ids is not None:32            filter_dict['id'] = set(ids).intersection(set(proxy.torrents.keys()))33        document = {34            'torrents': []35        }36        response = Deferred()37        deffered_torrents = proxy.get_torrents_status(filter_dict, filter_fields)38        def on_complete(torrents_dict):39            document['torrents'] = list(torrents_dict.values())40            response.callback(document)41        deffered_torrents.addCallback(on_complete)42        return response43    @export_api44    def add_torrent(self, metainfo, options=None):45        """Adds a torrent with the given options.46        metainfo could either be base64 torrent data or a magnet link.47        Returns `torrent_id` string or None.48        Available options are listed in deluge.core.torrent.TorrentOptions.49        :rtype: None|str50        """51        if options is None:52            options = {}53        coreconfig = component.get('CoreConfig')54        if 'download_location' not in options.keys():55            options.update({'download_location': coreconfig.get("download_location")})56        if common.is_magnet(metainfo):57            LOGGER.info('Adding torrent from magnet URI `%s` using options `%s` ...', metainfo, options)58            result = client.core.add_torrent_magnet(metainfo, options)59        else:60            LOGGER.info('Adding torrent from base64 string using options `%s` ...', options)61            result = client.core.add_torrent_file(None, metainfo, options)62        return result63    @export_api64    def remove_torrent(self, torrent_id, remove_data=False):65        """Removes a given torrent. Optionally can remove data."""66        return client.core.remove_torrent(torrent_id, remove_data)67    @export_api68    def get_api_version(self):69        """Returns WebAPI plugin version."""70        from webapi import VERSION...

test_export.py

Source:test_export.py

1import unittest2import pykintone3import tests.envs as envs4class TestExport(unittest.TestCase):5    def test_uesrs(self):6        export_api = pykintone.load(envs.FILE_PATH).user_api().for_exporting7        users = export_api.get_users().users8        self.assertTrue(len(users) > 0)9        for u in users:10            self.assertTrue(u.name)11    def test_user_organization_titles(self):12        export_api = pykintone.load(envs.FILE_PATH).user_api().for_exporting13        users = export_api.get_users().users14        tested = False15        for u in users[:5]:16            code = u.code17            ots_result = export_api.get_user_organization_titles(code)18            self.assertTrue(ots_result.ok)19            ots = ots_result.organization_titles20            if len(ots) > 0:21                tested = True22                for ot in ots:23                    self.assertTrue(ot.organization.code)24                    if ot.title is None:25                        print("ot.title is None. Please set a job title to the user {} for stricter test.".format(code))26                    else:27                        self.assertTrue(ot.title.code)28        if not tested:29            print("Caution, organization and title deserialization is not checked. Please make user who belongs to some organization.")30    def test_uesr_groups(self):31        from datetime import datetime32        export_api = pykintone.load(envs.FILE_PATH).user_api().for_exporting33        users = export_api.get_users().users34        tested = False35        for u in users[:5]:36            code = u.code37            groups_result = export_api.get_user_groups(code)38            self.assertTrue(groups_result.ok)39            if len(groups_result.groups) > 0:40                tested = True41                for g in groups_result.groups:42                    self.assertTrue(g.name)43        if not tested:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.