analitics

Pages

Monday, August 19, 2024

Python 3.12.1 : Web server with SQLite database using flask - update.

Update with new URL with params, see the first tutorial:
from flask import Flask, request, jsonify, render_template_string
import sqlite3
from datetime import datetime

app = Flask(__name__)

# Clasa pentru serverul SQL
class SQLiteServer:
    def __init__(self, db_name):
        self.db_name = db_name
        self.init_db()

    def init_db(self):
        conn = sqlite3.connect(self.db_name)
        c = conn.cursor()
        c.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                first_name TEXT,
                last_name TEXT,
                occupation TEXT,
                hobby TEXT,
                year_of_birth INTEGER,
                age INTEGER
            )
        ''')
        conn.commit()
        conn.close()

    def calculate_age(self, year_of_birth):
        # Adjust year_of_birth if only two digits are provided
        if len(str(year_of_birth)) == 2:
            if year_of_birth > int(str(datetime.now().year)[-2:]):
                year_of_birth += 1900
            else:
                year_of_birth += 2000
        current_year = datetime.now().year
        return current_year - year_of_birth

    def add_user(self, first_name, last_name, occupation, hobby, year_of_birth):
        age = self.calculate_age(year_of_birth)
        conn = sqlite3.connect(self.db_name)
        c = conn.cursor()
        c.execute('''
            INSERT INTO users (first_name, last_name, occupation, hobby, year_of_birth, age)
            VALUES (?, ?, ?, ?, ?, ?)
        ''', (first_name, last_name, occupation, hobby, year_of_birth, age))
        conn.commit()
        conn.close()

    def get_users(self):
        conn = sqlite3.connect(self.db_name)
        c = conn.cursor()
        c.execute("SELECT * FROM users")
        users = c.fetchall()
        conn.close()
        return users
    def get_users_jsonify():
        conn = sqlite3.connect('sqlite_database.db')
        c = conn.cursor()
        c.execute("SELECT * FROM users")
        users = c.fetchall()
        conn.close()
        return jsonify(users)

# Clasa pentru serverul web
class WebServer:
    def __init__(self, sqlite_server):
        self.sqlite_server = sqlite_server

    def run(self):
        app.run(debug=True)
    #   adauga user si buton de redirect la pagina users
    @app.route('/')
    def index():
        users = sqlite_server.get_users()
        return render_template_string('''
            <!DOCTYPE html>
            <html lang="en">
            <head>
                <meta charset="UTF-8">
                <meta name="viewport" content="width=device-width, initial-scale=1.0">
                <title>Flask website for testing cypress with sqlite</title>
            </head>
            <body>
                <h2>Add User</h2>
                <form action="/add_user" method="post">
                    First Name: <input type="text" name="first_name"><br>
                    Last Name: <input type="text" name="last_name"><br>
                    Occupation: <input type="text" name="occupation"><br>
                    Hobby: <input type="text" name="hobby"><br>
                    Year of Birth: <input type="text" name="year_of_birth"><br>
                    <input type="submit" value="Add User">
                </form>
                <a href="http://127.0.0.1:5000/users"><button type="button">Show Users</button></a>
            </body>
            </html>
        ''', users=users)
    @app.route('/add_user', methods=['POST'])
    def add_user():
        first_name = request.form['first_name']
        last_name = request.form['last_name']
        occupation = request.form['occupation']
        hobby = request.form['hobby']
        year_of_birth = int(request.form['year_of_birth'])
        sqlite_server.add_user(first_name, last_name, occupation, hobby, year_of_birth)
        return 'User added successfully! <a href="/">Go back</a>'
    @app.route('/users', methods=['GET'])
    def get_users():
        query_type = request.args.get('query_type', 'simple')
        
        conn = sqlite3.connect('sqlite_database.db')
        c = conn.cursor()
        
        try:
            c.execute('SELECT name FROM sqlite_master WHERE type="table" AND name="users"')
            if not c.fetchone():
                return jsonify({"error": "Table 'users' does not exist"})

            if query_type == 'advanced':
                # Advanced query logic
                first_name = request.args.get('first_name')
                last_name = request.args.get('last_name')
                occupation = request.args.get('occupation')
                hobby = request.args.get('hobby')
                year_of_birth = request.args.get('year_of_birth')

                query = 'SELECT * FROM users WHERE 1=1'
                params = []
                # Exemple query simple 
                # Basic query: /users
                # Simple query: /users?query_type=simple for simple selection
                # Addvanced query: /users?query_type=advanced&first_name=John&occupation=Engineer for advanced querying
                # Advanced query with name search: /users?query_type=advanced&first_name=John&last_name=Doe
                # Query by occupation: /users?query_type=advanced&occupation=Engineer
                # Query by hobby: /users?query_type=advanced&hobby=Reading
                # Query by year of birth: /users?query_type=advanced&year_of_birth=1990

                if first_name:
                    query += ' AND first_name LIKE ?'
                    params.append(f'%{first_name}%')
                if last_name:
                    query += ' AND last_name LIKE ?'
                    params.append(f'%{last_name}%')
                if occupation:
                    query += ' AND occupation LIKE ?'
                    params.append(f'%{occupation}%')
                if hobby:
                    query += ' AND hobby LIKE ?'
                    params.append(f'%{hobby}%')
                if year_of_birth:
                    query += ' AND year_of_birth = ?'
                    params.append(year_of_birth)

                # Query by minimum age: /users?query_type=advanced&min_age=30
                # Query by maximum age: /users?query_type=advanced&max_age=50
                # Query with ordering: /users?query_type=advanced&order_by=last_name
                # Query with limit: /users?query_type=advanced&limit=10
                # Combined query: /users?query_type=advanced&first_name=John&occupation=Engineer&min_age=25&order_by=year_of_birth&limit=5          
                # Additional advanced query options
                for param, value in request.args.items():
                    match param:
                        case 'min_age':
                            query += ' AND (? - year_of_birth) >= ?'
                            params.extend([datetime.now().year, int(value)])
                        case 'max_age':
                            query += ' AND (? - year_of_birth) <= ?'
                            params.extend([datetime.now().year, int(value)])
                        case 'order_by':
                            query += f' ORDER BY {value}'
                        case 'limit':
                            query += ' LIMIT ?'
                            params.append(int(value))
                c.execute(query, params)
            else:
                # Simple query logic
                c.execute('SELECT * FROM users')

            users = c.fetchall()
        except sqlite3.OperationalError as e:
            return jsonify({"error": str(e)})
        finally:
            conn.close()
        
        return jsonify(users)

# Instanțierea serverului SQL și a serverului web
sqlite_server = SQLiteServer('sqlite_database.db')
web_server = WebServer(sqlite_server)

if __name__ == '__main__':
    web_server.run()