Tag: Python

I’m fairly new to time series databases in general and Influx in particular.
My objetive is to build a simple and general use API that will allow me to write and read data from an Influx Database.
I used Influx on a previous project and I found that building the API for the database was pretty mechanical, so I decided to create a general API.

I divided my code in 3 modules:

• Main.py where I will run FastAPI
• Database.py where I connect with the database and I set the writing and reading actions.
• Model.py where I create the pydantic model for the API.

Main.py

from fastapi import FastAPI
import uvicorn
import yaml
from yaml.loader import SafeLoader
from Model.Model import WritingData, ReadingData
from Database.Database import InfluxDataBase



with open("config.yaml", "r") as ymlfile:
    cfg = yaml.load(ymlfile,Loader=SafeLoader)
server_URL=cfg("InfluxDB")("server_URL")
token=cfg("InfluxDB")("token")
org=cfg("InfluxDB")("org")

Influx = InfluxDataBase(server_URL,token,org)
app = FastAPI()

@app.post('/write/')
async def call_writing_influx(data: WritingData):
    Influx.write_data(data)

@app.post('/read/')
async def call_reading_influx(data: ReadingData):
    return Influx.read_data(data)
    
if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=5000, reload=True)

Database

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import json

class InfluxDataBase:
    
    def __init__(self,server_URL,token,org) -> None:
        self.client=InfluxDBClient(server_URL, token=token, org=org)
        self.write_api=self.client.write_api(write_options=SYNCHRONOUS)
        self.query_api=self.client.query_api()
        self.server_URL=server_URL
        self.token=token
        self.org=org

    def write_data(self,data) -> None:
        executable_code='Point(data.measurement)'
        n_fields=len(data.field)
        n_tag=len(data.tag)
        for i in range(n_tag):
            executable_code=executable_code+'.tag(list(data.tag.keys())({}),list(data.tag.values())({}))'.format(i,i)
        for i in range(n_fields):
            executable_code=executable_code+'.field(list(data.field.keys())({}),list(data.field.values())({}))'.format(i,i)
        if data.timestamp is not None:
            executable_code=executable_code+'.time(data.timestamp)'

        Data=eval(executable_code)
        self.write_api.write(bucket=data.bucket_name, record=Data)
    
    def read_data(self,data):
        query = f'''
        from(bucket: "{data.bucket_name}")'''+ ''' 
        |> range(start: -{}h, stop: now())'''.format(data.time_interval)+f'''
        |> filter(fn:(r) => r("_measurement") == "{data.measureament_name}")'''
        for i in range(len(data.tag)):
            query=query+f'''|> filter(fn:(r) => r("{list(data.tag.keys())(i)}") == "{list(data.tag.values())(i)}") '''
        for i in range(len(data.field)):
            query=query+f'''|> filter(fn:(r) => r("_field") == "{data.field(i)}") '''
        result = self.query_api.query(org=self.org, query=query) 
        results={}
        for table in result:
            for record in table.records:
                results(record.get_field())=record.get_value()
        json_result=json.dumps(results)
        return json_result

Model.py

from pydantic import BaseModel
from typing import List,Dict,Optional
from datetime import datetime

class WritingData(BaseModel):
    bucket_name: str
    measurement: str
    tag: Dict(str,str)
    field: Dict(str,float)
    timestamp: Optional(datetime) = None

class ReadingData(BaseModel):
    bucket_name: str
    time_interval: int
    measureament_name: str
    tag: Dict(str,str)
    field: List(str)

My current code works fine, It does what is intended to do.

There is any good practice politics I’m breaking?

Do you have any suggestions to imptove it?