Issue:

Pandas/NumPy are not required for Influx interaction, but are dependencies for aioinflux.

When developing for a Raspberry target, this becomes an issue, as Pandas/NumPy do not provide compiled packages for ARMv7. Compiling these packages on a Raspberry 3 takes +/- 1 hour. That's a bit much for an unused dependency.

Desired behavior:

No functional changes for clients that use dataframe serialization functionality. No functional changes for clients that don't - but they can drop the Pandas/Numpy packages from their dependency stack.

Proposed solution:

Pandas/NumPy dependencies in setup.py can move to an extras_require collection.

client.py does not use NumPy, and only uses Pandas to define PointType.

• The Pandas import can be contained inside a try/except block.
• The PointType definition can equally easily be made conditional.

serialization.py makes more extensive use of dependencies.

• make_df() and parse_df() are Pandas-only functions, and can move to a conditional include.
• The isinstance(data, pd.DataFrame) check can also be made conditional (if using_pd and isinstance(data, pd.DataFrame).
• Same for type checks in _parse_fields(): the checks for np.integer and np.isnan() can be placed behind evaluation of a using_np variable, or pd is None check.

Required effort:

Between 2 hours and 2 days.

Practical consideration:

We're actively using aioinflux (apart from the compile time issues it works great), and I can make the time to make a PR. Bigger issue is whether this is a desired feature for the main repository. If not, I can fork and implement it downstream.

Hi, thanks again for this great package, been very helpful.

Sync InfluxDB client has in its constructor a path parameter:

https://influxdb-python.readthedocs.io/en/latest/api-documentation.html#influxdb.InfluxDBClient

path (str) – path of InfluxDB on the server to connect, defaults to ‘’

And the URL is built as follows:

https://github.com/influxdata/influxdb-python/blob/d5d12499f3755199d5eedd8b363450f1cf4073bd/influxdb/client.py#L123

        self.__baseurl = "{0}://{1}:{2}{3}".format(
            self._scheme,
            self._host,
            self._port,
            self._path)

Although in aioinflux there is no path parameter, and the URL is built as follows:

https://github.com/gusutabopb/aioinflux/blob/master/aioinflux/client.py#L163

        return f'{"https" if self.ssl else "http"}://{self.host}:{self.port}/{{endpoint}}'

So, it seems that I cannot connect with aioinflux to our Influx deployment, as for reasons unknown to me, it is under a path.

Currently, I created a quick monkey patch as follows:

class MonkeyPatchedInfluxDBClient(InfluxDBClient):
    def __init__(self, *args, path='/', **kwargs):
        super().__init__(*args, **kwargs)
        self._path = path

    @property
    def path(self):
        return self._path

    @property
    def url(self):
        return '{protocol}://{host}:{port}{path}{{endpoint}}'.format(
            protocol='https' if self.ssl else 'http',
            host=self.host,
            port=self.port,
            path=self.path,
        )

Thanks for placing the url in a property, that was useful.

When the query has group by clause other than time, for example

SELECT COUNT(*) FROM "db"."rp"."measurement" WHERE time > now() - 7d GROUP BY "category"

The dataframe output mode returns a dictionary instead of dataframe. The key seems to be a string with "measurement_name, category=A", "measurement_name, category=B",... and values of the dictionary are dataframes. Is this expected?

Hi, Gustavo,

I tried to follow the demo on our project which doesnt work, could you help me to figure out the reason?

here is my code

async def read_influxdb(userid, starttime, endtime):
    #logger = logging.getLogger("influxDB read demo")
    async with InfluxDBClient(host=localhost, port=8086, username='admin', password='123456',db=db_name) as client:
        user_id = '\'' + str(userid) + '\''
        sql_ecg = 'SELECT point FROM wave WHERE (person_zid = {}) AND (time > {}s) AND (time < {}s)'.format(user_id, starttime, endtime)
        await client.query(sql_ecg, chunked=True)

if __name__ ==  '__main__':
    user_id = 973097
    starttime = '2018-09-26 18:08:48'
    endtime = '2018-09-27 18:08:48'
    starttime_posix = utc_to_local(starttime)
    endtime_posix = utc_to_local(endtime)
    asyncio.get_event_loop().run_until_complete(read_influxdb(user_id, starttime_posix, endtime_posix))

We I run this code, I get the errors below:

sys:1: RuntimeWarning: coroutine 'query' was never awaited
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x10f78f630>

Best

Aioinflux used to provide a built-in caching local functionality using Redis. However, due to low perceived usage, vendor lock-in (Redis) and extra complexity added to Aioinflux, I have decided to remove it.

Hopefully no one else besides my past self use this functionality. In case someone else did, or in case someone else didn't but may be interested in caching InfluxDB query results, I will add a simple implementation of a simple caching layer using pickle. If this affects you please let me know by commenting below.

Background

PEP 563 behavior is available from Python 3.7 (using from __future__ import annotations) and will become the default in Python 3.10.

Description of the problem

Among changes introduced by PEP 563, the type annotations in __annotations__ attribute of an object are stored in string form. This breaks in the function below because all the tests expect type objects. https://github.com/gusutabopb/aioinflux/blob/77f9d24f493365356298a1eb904a27ce046cec27/aioinflux/serialization/usertype.py#L57-L67

Reproduction

• Define a user-defined class, decorated with lineprotocol():

from typing import NamedTuple

import aioinflux


@aioinflux.lineprotocol
class Production(NamedTuple):
    total_line: aioinflux.INT

# Works well as is

• Add from __future__ import annotations at the top and you get: SchemaError: Must have one or more non-empty field-type attributes [~BOOL, ~INT, ~DECIMAL, ~FLOAT, ~STR, ~ENUM] at import time.

Possible solution

Using https://docs.python.org/3/library/typing.html#typing.get_type_hints has the same behavior (returns a dict with values as type objects) with or without from __future__ import annotations. Furthermore, the autor of PEP 563 advises to use it.

Hi

During processing this query:

SELECT ROUND(LAST(Free_Megabytes) / 1024) AS free, ROUND(Free_Megabytes / 1024 / (Percent_Free_Space / 100)) AS total, ROUND(Free_Megabytes / 1024 * ((100 - Percent_Free_Space) / Percent_Free_Space)) AS used, (100 - Percent_Free_Space) as percent, instance as path FROM win_disk WHERE host = 'ais-pc-16003' GROUP BY instance

This is the raw data that InfluxDBClient.query returned.

{'results': [{'series': [{'columns': ['time',
                                      'free',
                                      'total',
                                      'used',
                                      'percent',
                                      'path'],
                          'name': 'win_disk',
                          'tags': {'instance': 'C:'},
                          'values': [[1577419571000000000,
                                      94,
                                      238,
                                      144,
                                      60.49140930175781,
                                      'C:']]},
                         {'columns': ['time',
                                      'free',
                                      'total',
                                      'used',
                                      'percent',
                                      'path'],
                          'name': 'win_disk',
                          'tags': {'instance': 'D:'},
                          'values': [[1577419571000000000,
                                      1727,
                                      1863,
                                      136,
                                      7.3103790283203125,
                                      'D:']]},
                         {'columns': ['time',
                                      'free',
                                      'total',
                                      'used',
                                      'percent',
                                      'path'],
                          'name': 'win_disk',
                          'tags': {'instance': 'HarddiskVolume1'},
                          'values': [[1577419330000000000,
                                      0,
                                      0,
                                      0,
                                      29.292930603027344,
                                      'HarddiskVolume1']]},
                         {'columns': ['time',
                                      'free',
                                      'total',
                                      'used',
                                      'percent',
                                      'path'],
                          'name': 'win_disk',
                          'tags': {'instance': '_Total'},
                          'values': [[1577419571000000000,
                                      1821,
                                      2101,
                                      280,
                                      13.345237731933594,
                                      '_Total']]}],
              'statement_id': 0}]}

And I want to use this code to get parsed dicts:

def dict_parser(*x, meta):
    return dict(zip(meta['columns'], x))

g = fixed_iterpoints(r, dict_parser)

But only got the first row ("instance": "C:"). And below is the source of iterpoints. As you can see, the for-loop returned at the first iteration.

def iterpoints(resp: dict, parser: Optional[Callable] = None) -> Iterator[Any]:
    for statement in resp['results']:
        if 'series' not in statement:
            continue
        for series in statement['series']:
            if parser is None:
                return (x for x in series['values'])
            elif 'meta' in inspect.signature(parser).parameters:
                meta = {k: series[k] for k in series if k != 'values'}
                meta['statement_id'] = statement['statement_id']
                return (parser(*x, meta=meta) for x in series['values'])
            else:
                return (parser(*x) for x in series['values'])
    return iter([])

I modified this function as a workaround:

def fixed_iterpoints(resp: dict, parser: Optional[Callable] = None):
    for statement in resp['results']:
        if 'series' not in statement:
            continue

        gs = []
        for series in statement['series']:
            if parser is None:
                part = (x for x in series['values'])
            elif 'meta' in inspect.signature(parser).parameters:
                meta = {k: series[k] for k in series if k != 'values'}
                meta['statement_id'] = statement['statement_id']
                part = (parser(x, meta=meta) for x in series['values'])
            else:
                part = (parser(x) for x in series['values'])

            if len(statement['series']) == 1:
                return part

            gs.append(part)

        return gs
    return iter([])

It worked for me. But it returned nested generator which might be wierd. I want to know if you have a better idea.

When adding extra_tags to a user defined object the extra tags are not properly escaped to line protocol. This PR ensures the tag value is escaped, reusing the existing tag_escape implementation. I did not alter the tag name but I'd be fine adding that in as well if requested.

First of, thank you! Great repo with excellent documentation. I use it with a Starlette project I am working on.

In the project I've implemented a simple way to parse a pandas.Dataframe from a chuncked response. It works, and I added it to my fork and I am wondering if you would welcome such a feature.

Here is the MVP implementation in my fork

I'll clean the code, remove exceptions, move it to serialization/dataframe.py and add tests if you're OK with it.

Currently, the blocking mode won't work on Python 3.7 running on Jupyter. The code below:

import aioinflux
c = aioinflux.InfluxDBClient(db='mydb', mode='blocking')
c.show_measurements()

Raises RuntimeError: This event loop is already running

This is caused by the fact that the latest versions of Tornado (which is used by Jupyter/ipykernel) runs an asyncio loop on the main thread by default:

# Python 3.7
import asyncio
asyncio.get_event_loop()
# <_UnixSelectorEventLoop running=True closed=False debug=False>

# Python 3.6 (w/ tornado < 5)
import asyncio
asyncio.get_event_loop()
# <_UnixSelectorEventLoop running=False closed=False debug=False>

This is being discussed on https://github.com/jupyter/notebook/issues/3397

From an aioinflux perspective, a possible work around would be to start a new event loop on a background thread and use asyncio.run_coroutine_threadsafe to run the coroutine and return a concurrent.futures.Future object that wraps the result.

Thanks for creating this library? Does it support UDP inserts via asyncio?

i.e. udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

udp_port_tuple = (host, udp_port) udp_socket.sendto(data_str, udp_port_tuple)

I have no intention of using pandas and don't want to see the following warning when my application starts or worry about suppressing a warning:

"Pandas/Numpy is not available. Support for 'dataframe' mode is disabled."

Please consider either removing this warning on import.

When trying to write an integer64 field, I was getting an error due to the presence of missing values. The missing values were in the form of pd.NA, rather than np.nan and they were not being excluded in the serialization.

I made an attempt to fix this and it worked, though might not be the most elegant solution. In the _replace function, I added a new replacement tuple to the list of replacements, very similar to the one that handles the nans:

def _replace(df):
    obj_cols = {k for k, v in dict(df.dtypes).items() if v is np.dtype('O')}
    other_cols = set(df.columns) - obj_cols
    obj_nans = (f'{k}="nan"' for k in obj_cols)
    other_nans = (f'{k}=nani?' for k in other_cols)
    obj_nas = (f'{k}="<NA>"' for k in obj_cols)
    other_nas = (f'{k}=<NA>i?' for k in other_cols)
    replacements = [
        ('|'.join(chain(obj_nans, other_nans)), ''),
        ('|'.join(chain(obj_nas, other_nas)), ''),
        (',{2,}', ','),
        ('|'.join([', ,', ', ', ' ,']), ' '),
    ]
    return replacements

Hope this ends up helping someone

Hi, If you find the time for some maintenance could you include the LICENSE file in the next pypi release? This simplifies integration of the package through yocto/bitbake into embedded linux applications. Best regards

datetime objects were handled incorrectly. This resulted in a time offset from UTC.

This correct implementation assumes UTC time, if no tzinfo object is attached to the datetime. Further the offset is now taken from the tzinfo object.

Just wondering if this library is still actively maintained since it hasn't had a commit or merged PR since last summer. No judgment, just wondering since I like the idea of this client vs influx-python.