from flask import Flask, jsonify, request from flask_sqlalchemy import SQLAlchemy from datetime import datetime import uuid app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///tokens.db' # Use SQLite for simplicity app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False db = SQLAlchemy(app) class Token(db.Model): id = db.Column(db.String(36), primary_key=True, default=str(uuid.uuid4())) email = db.Column(db.String(255), nullable=False) access_token = db.Column(db.String(255), nullable=False) refresh_token = db.Column(db.String(255), nullable=False) expiration_seconds = db.Column(db.Integer, nullable=False) readwise_api_key = db.Column(db.String(255)) active = db.Column(db.Boolean, default=True) created_at = db.Column(db.DateTime, default=datetime.utcnow) updated_at = db.Column(db.DateTime, default=datetime.utcnow) def __repr__(self): return f'' # Create an application context with app.app_context(): db.create_all() # API to create a new token entry @app.route('/token', methods=['POST']) def create_token(): data = request.get_json() email = data.get('email') access_token = data.get('access_token') refresh_token = data.get('refresh_token') expiration_seconds = data.get('expiration_seconds') required_fields = ['email', 'access_token', 'refresh_token', 'expiration_seconds'] missing_fields = [field for field in required_fields if not data.get(field)] if missing_fields: return jsonify({'error': f'Missing required fields: {", ".join(missing_fields)}'}), 400 # unique email when active is true existing_token = Token.query.filter_by(email=email, active=True).first() if existing_token: return jsonify({'error': 'An active token with this email already exists'}), 400 new_token = Token(email=email, access_token=access_token, refresh_token=refresh_token, expiration_seconds=expiration_seconds) db.session.add(new_token) db.session.commit() return jsonify({'id': new_token.id}), 201 # API to get the token based on the id @app.route('/token/', methods=['GET']) def get_token_by_id(id): token = Token.query.get(id) if not token: return jsonify({'error': 'Token not found'}), 404 token_info = { 'id': token.id, 'email': token.email, 'access_token': token.access_token, 'refresh_token': token.refresh_token, 'expiration_seconds': int(token.expiration_seconds), 'readwise_api_key': token.readwise_api_key, 'active': token.active, 'created_at': int(token.created_at.timestamp()), 'updated_at': int(token.updated_at.timestamp()) } return jsonify({'token': token_info}), 200 # API to get the token based on the email @app.route('/token', methods=['GET']) def get_token_by_email(): email = request.args.get('email') if not email: return jsonify({'error': 'Missing email query parameter'}), 400 token = Token.query.filter_by(email=email, active=True).first() if not token: return '', 204 token_info = { 'id': token.id, 'email': token.email, 'access_token': token.access_token, 'refresh_token': token.refresh_token, 'expiration_seconds': int(token.expiration_seconds), 'readwise_api_key': token.readwise_api_key, 'active': token.active, 'created_at': int(token.created_at.timestamp()), 'updated_at': int(token.updated_at.timestamp()) } return jsonify({'token': token_info}), 200 # API to update the token based on the id @app.route('/token/', methods=['PUT']) def update_token_by_id(id): token = Token.query.get_or_404(id) data = request.get_json() token.access_token = data.get('access_token', token.access_token) token.refresh_token = data.get('refresh_token', token.refresh_token) token.expiration_seconds = data.get('expiration_seconds', token.expiration_seconds) token.active = data.get('active', token.active) token.readwise_api_key = data.get('readwise_api_key', token.readwise_api_key) token.updated_at = datetime.utcnow() db.session.commit() return '', 204 # get all tokens @app.route('/token/all', methods=['GET']) def get_all_tokens(): tokens = Token.query.all() tokens_info = [{ 'id': token.id, 'email': token.email, 'access_token': token.access_token, 'refresh_token': token.refresh_token, 'expiration_seconds': int(token.expiration_seconds), 'readwise_api_key': token.readwise_api_key, 'active': token.active, 'created_at': int(token.created_at.timestamp()), 'updated_at': int(token.updated_at.timestamp()) } for token in tokens] return jsonify({'tokens': tokens_info}), 200 if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=True)