<commit_before>#!/usr/bin/env python3
# https://www.tutorialspoint.com/sqlalchemy/sqlalchemy_core_creating_table.htm

import datetime
import json

import dateutil.parser
import paho.mqtt.client as mqtt
import pandas as pd
from sqlalchemy import create_engine, MetaData, Table, Column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, BigInteger, DateTime
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///mqtt_test.sqlite', echo=True)
meta = MetaData()

# Base = declarative_base()
# Session = sessionmaker(bind=engine)
# session = Session()
# Base.metadata.create_all(engine)

message = Table('message', meta,
                Column('id', BigInteger, primary_key=True),
                Column('rectime', String),
                Column('sendtime', String),
                Column('message', String))
meta.create_all(engine)


# class Message(Base):
#     __tablename__ = 'message'
#     id = Column(BigInteger, primary_key=True)
#     rectime = Column(String(26))
#     sendtime = Column(String(26))
#     message = Column(String)
#
#     def __repr__(self):
#         return "<Message(id={}, rectime={}, sendtime={}, message={}".format(self.id, self.rectime, self.sendtime,
#                                                                             self.message)
#

# This is the Subscriber

def on_connect(client, userdata, flags, rc):
    print("Connected with result code " + str(rc))
    client.subscribe("topic/test")


def _msg_to_db(msg, rectime):
    print(rectime.isoformat(), msg['datetime'], msg['Message'])
    ins = message.insert().values(rectime=rectime.isoformat(), sendtime=msg['datetime'],
                                  message=msg['Message'])
    conn = engine.connect()
    result = conn.execute(ins)
    print("Result", result)
    # row_Message = Message(rectime=rectime,
    #                       sendtime=dateutil.parser.parse(msg['datetime']),
    #                       message=msg['Message'])
    # session.add(row_Message)
    # session.commit()


def on_message(client, userdata, msg):
    rectime = datetime.datetime.now()
    message = msg.payload.decode()
    message_val = json.loads(message)
    _msg_to_db(message_val, rectime)


client = mqtt.Client()
client.connect("localhost", 1883, 60)

client.on_connect = on_connect
client.on_message = on_message

client.loop_forever()
<commit_msg>working demo with pandas, less streamlined but works<commit_after>#!/usr/bin/env python3
# https://www.tutorialspoint.com/sqlalchemy/sqlalchemy_core_creating_table.htm

import datetime
import json

import dateutil.parser
import paho.mqtt.client as mqtt
import pandas as pd
# from sqlalchemy import create_engine, MetaData, Table, Column
# from sqlalchemy.ext.declarative import declarative_base
# from sqlalchemy import Column, Integer, String, BigInteger, DateTime
# from sqlalchemy.orm import sessionmaker
#
# engine = create_engine('sqlite:///mqtt_test.sqlite', echo=True)
# meta = MetaData()
#
# # Base = declarative_base()
# # Session = sessionmaker(bind=engine)
# # session = Session()
# # Base.metadata.create_all(engine)
#
# message = Table('message', meta,
#                 Column('id', BigInteger, primary_key=True),
#                 Column('rectime', String),
#                 Column('sendtime', String),
#                 Column('message', String))
# meta.create_all(engine)


# class Message(Base):
#     __tablename__ = 'message'
#     id = Column(BigInteger, primary_key=True)
#     rectime = Column(String(26))
#     sendtime = Column(String(26))
#     message = Column(String)
#
#     def __repr__(self):
#         return "<Message(id={}, rectime={}, sendtime={}, message={}".format(self.id, self.rectime, self.sendtime,
#                                                                             self.message)
#

from sqlalchemy import create_engine

engine = create_engine('sqlite:///mqtt_test.sqlite', echo=True)


# This is the Subscriber

def on_connect(client, userdata, flags, rc):
    print("Connected with result code " + str(rc))
    client.subscribe("topic/test")


def _msg_to_db(msg, rectime):
    print(rectime, msg['datetime'], msg['Message'])
    df = pd.DataFrame({'rectime': rectime, 'message': msg['Message']}, index=pd.DatetimeIndex([msg['datetime']]))
    print(df)
    df.to_sql('message', con=engine, if_exists='append')
    # ins = message.insert().values(rectime=rectime.isoformat(), sendtime=msg['datetime'],
    #                               message=msg['Message'])
    # conn = engine.connect()
    # result = conn.execute(ins)
    # print("Result", result)
    # # row_Message = Message(rectime=rectime,
    #                       sendtime=dateutil.parser.parse(msg['datetime']),
    #                       message=msg['Message'])
    # session.add(row_Message)
    # session.commit()


def on_message(client, userdata, msg):
    rectime = datetime.datetime.now()
    message = msg.payload.decode()
    message_val = json.loads(message)
    _msg_to_db(message_val, rectime)


client = mqtt.Client()
client.connect("localhost", 1883, 60)

client.on_connect = on_connect
client.on_message = on_message

client.loop_forever()

{'fraction_non_alphanumeric': 0.13441170806855382, 'fraction_numerical': 0.004236472174080493, 'mean_word_length': 3.2713815789473686, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2329156', 'n_tokens_mistral': 1610, 'n_tokens_neox': 1557, 'n_words': 318}