chatgpt/webapp/test.py
2022-12-17 22:39:21 +02:00

66 lines
1.9 KiB
Python
Executable File

#!/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()