initial save of my chatGPT coding

webapp/celery-test.py

@@ -0,0 +1,37 @@
+#!/usr/bin/python3
+
+from celery import Celery
+from celery.backends import sqlalchemy
+from flask import Flask
+import requests
+
+# create Flask app
+app = Flask(__name__)
+
+# create a Celery app
+celery_app = Celery('tasks')
+
+# configure the app to use the sqlalchemy backend and specify the SQLite database URL as the broker
+app.conf.update(
+    result_backend='sqlalchemy',
+    broker_url='sqlite:///celery.db',
+    task_serializer='json',
+    result_serializer='json',
+)
+
+@celery_app.task
+def get_public_ip():
+    response = requests.get('https://ifconfig.me/ip')
+    return response.text
+
+# define a Flask route that returns the current public IP address
+@app.route('/public_ip')
+def public_ip():
+    # call the get_public_ip task and wait for it to complete
+    result = get_public_ip.delay().get()
+
+    # return the task result as the response
+    return result
+
+if __name__ == '__main__':
+    app.run()

webapp/flask-dump.py

@@ -0,0 +1,70 @@
+#!/usr/bin/python3
+import json
+import sqlite3
+import requests
+import time 
+from flask import Flask
+
+# Initialize the Flask app
+app = Flask(__name__)
+
+# Dump the contents of a SQLite database to the response
+@app.route("/dump/<ask_timestamp>")
+def dump(ask_timestamp):
+    # Open the database connection
+    con = sqlite3.connect("../btc_timeseries.db")
+
+
+    # Create a cursor to navigate the database
+    cur = con.cursor()
+
+    # Fetch all rows from the table
+    rows = cur.execute("SELECT * FROM timeseries").fetchall()
+
+    data = {
+            "parameter": ask_timestamp,
+            "rows": rows
+            }
+
+    # Build the response as a string
+    response = json.dumps(data)
+
+    con.close()
+
+    for row in rows:
+        old_timestamp = time.strptime(row[0], "%Y-%m-%d %H:%M:%S")
+        unix_timestamp = time.mktime(old_timestamp)
+        if int(ask_timestamp) < int(unix_timestamp):
+            print('EQUALS: ', ask_timestamp, ' AND ', unix_timestamp)
+        else:
+            print('NOPE ', row[0], ' AS ', unix_timestamp)
+
+    return response
+
+#@app.route("/dump/timestamp")
+#def dump(timestamp):
+#    # Open the database connection
+#    con = sqlite3.connect("../btc_timeseries.db")
+#
+#    print(timestamp)
+#
+#    # Create a cursor to navigate the database
+#    cur = con.cursor()
+#
+#    # Fetch all rows from the table
+#    rows = cur.execute("SELECT * FROM timeseries").fetchall()
+#
+#    # Build the response as a string
+#    response = ""
+#    for row in rows:
+#        response += ", ".join(row) + "\n"
+#
+#    # Close the database connection
+#    con.close()
+#
+#    # Return the response
+#    return response
+
+# Run the app
+if __name__ == "__main__":
+    app.run()

webapp/test.py

@@ -0,0 +1,65 @@
+#!/usr/bin/python3
+
+from celery import Celery
+from celery.backends import sqlalchemy
+from datetime import datetime
+from flask import Flask
+import requests
+from sqlalchemy import create_engine, Column, Integer, String, DateTime
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+# create a Flask app
+app = Flask(__name__)
+
+# create a Celery app and configure it to use the sqlalchemy backend and the SQLite database specified by the broker_url
+celery_app = Celery('tasks')
+celery_app.conf.update(
+    result_backend='sqlalchemy',
+    broker_url='sqlite:///celery.db',
+    task_serializer='json',
+    result_serializer='json',
+)
+
+# define the get_public_ip Celery task (as in the previous example)
+@celery_app.task
+def get_public_ip():
+    response = requests.get('https://ifconfig.me/ip')
+    return response.text
+
+# create an SQLAlchemy base class
+Base = declarative_base()
+
+# define an IPHistory model that will be used to store IP addresses and timestamps in the database
+class IPHistory(Base):
+    __tablename__ = 'ip_history'
+
+    id = Column(Integer, primary_key=True)
+    ip_address = Column(String)
+    timestamp = Column(DateTime, default=datetime.now)
+
+# create a database engine and connect to the ip_history.db database
+engine = create_engine('sqlite:///ip_history.db')
+Base.metadata.create_all(engine)
+
+# create a session object that will be used to add records to the database
+Session = sessionmaker(bind=engine)
+session = Session()
+
+# define a Flask route that returns the current public IP address
+@app.route('/public_ip')
+def public_ip():
+    # call the get_public_ip task and wait for it to complete
+    ip_address = get_public_ip.delay().get()
+
+    # create a new IPHistory record and add it to the database
+    record = IPHistory(ip_address=ip_address)
+    session.add(record)
+    session.commit()
+
+    # return the current IP address as the response
+    return ip_address
+
+if __name__ == '__main__':
+    app.run()
+