Initial commit

.gitignore

@@ -0,0 +1,3 @@
+__pychache__
+config.toml
+database.db

README.md

@@ -0,0 +1,10 @@
+# PyPolls
+
+## API endpoints
+Description                   | Method   | Endpoint                                   
+-----------                   | :-------: | ---------------------------------------
+Create new poll               | POST      | /api/v1/polls/create/<choice/checkboxes>
+Vote                          | POST, PUT | /api/v1/polls/<uuid:poll_id>?id=...<br />/api/v1/polls/<uuid:poll_id>?0=true&1=false&2=true
+Get results and variants      | GET       | /api/v1/polls/<uuid:poll_id>           
+Stop poll                     | POST, PUT | /api/v1/stop/<uuid:poll_id>[?token=...] [Authorization: Bearer ...]
+Delete poll                   | DELETE    | /api/v1/polls/<uuid:poll_id>[?token=...] [Authorization: Bearer ...]

config.example.toml

@@ -0,0 +1,2 @@
+[database]
+file = "/path/to/database.db"

flake.lock

@@ -0,0 +1,59 @@
+{
+  "nodes": {
+    "flake-utils": {
+      "inputs": {
+        "systems": "systems"
+      },
+      "locked": {
+        "lastModified": 1685518550,
+        "narHash": "sha256-o2d0KcvaXzTrPRIo0kOLV0/QXHhDQ5DTi+OxcjO8xqY=",
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "rev": "a1720a10a6cfe8234c0e93907ffe81be440f4cef",
+        "type": "github"
+      },
+      "original": {
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "type": "github"
+      }
+    },
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1685399834,
+        "narHash": "sha256-Lt7//5snriXSdJo5hlVcDkpERL1piiih0UXIz1RUcC4=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "58c85835512b0db938600b6fe13cc3e3dc4b364e",
+        "type": "github"
+      },
+      "original": {
+        "id": "nixpkgs",
+        "type": "indirect"
+      }
+    },
+    "root": {
+      "inputs": {
+        "flake-utils": "flake-utils",
+        "nixpkgs": "nixpkgs"
+      }
+    },
+    "systems": {
+      "locked": {
+        "lastModified": 1681028828,
+        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
+        "owner": "nix-systems",
+        "repo": "default",
+        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-systems",
+        "repo": "default",
+        "type": "github"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

requirements.txt

@@ -0,0 +1,2 @@
+Flask==2.0.2
+tomli==2.0.1

shell.nix

@@ -0,0 +1,5 @@
+{ pkgs ? import <nixpkgs> {} }:
+  pkgs.mkShell {
+    # nativeBuildInputs is usually what you want -- tools you need to run
+    nativeBuildInputs = [ pkgs.buildPackages.python311 pkgs.buildPackages.python311Packages.flask pkgs.buildPackages.python311Packages.tomli pkgs.sqlite ];
+}

src/.gitignore

@@ -0,0 +1,4 @@
+config.toml
+database.db
+tr(uuid.uuid4())
+__pycache__

src/app.py

@@ -0,0 +1,265 @@
+from re import A
+from flask import (
+    Flask,
+    abort,
+    request,
+    jsonify,
+    render_template,
+    make_response,
+    Response,
+)
+import tomli
+import uuid
+import json
+import random
+import string
+from db import *
+
+protocol = "http"
+domain = "localhost:5000"
+
+with open("config.toml", mode="rb") as fp:
+    cfg = tomli.load(fp)
+
+initdb(db())
+
+app = Flask(__name__)
+
+# APIs
+def get_poll_type(n: int):
+    if n == 1 or n == 2:
+        return("choice")
+    elif n == 3 or n == 9:
+        return("checkboxes")
+    else:
+        return("unknown")
+
+def create(title, creator, description, entrs, polltype, alias = ""):
+    poll_id = str(uuid.uuid4())
+    admin_token = ''.join(random.choices(string.ascii_letters + string.digits, k=24))
+    if creator == None:
+        creator = ""
+    if description == None:
+        description = ""
+    conn = db()
+    if polltype == "checkboxes":
+        polltype = "3"
+    elif polltype == "choice":
+        polltype = "1"
+    for i in entrs:
+        conn.execute(f"INSERT INTO variants VALUES ('{poll_id}', '{i}', 0)")
+    conn.execute(
+        f"INSERT INTO polls VALUES ('{poll_id}', '{admin_token}', '{title}', '{creator}', '{description}', {polltype}, '{alias}', 0)"
+    )
+    conn.commit()
+    conn.close()
+    return (poll_id, admin_token)
+
+
+@app.route("/api/v1/polls/create/<string:poll_type>", methods=["POST"])
+def api_create(poll_type):
+    if poll_type != "checkboxes" and poll_type != "choice":
+        return '{"error":"Uknown poll type"}', 400
+
+    if request.is_json == False:
+        req = request.form.to_dict()
+    else:
+        req = request.get_json()
+    if req.get("title") and req.get("values"):
+        ret = create(req["title"], req.get("author"), req.get("description"), req["values"], poll_type)
+        return '{"id":"%s","token":"%s"}' % (ret[0], ret[1]), 200
+    else:
+        return '{"error":"Title and values are required"}', 400
+
+
+@app.route("/api/v1/polls/<uuid:poll_id>", methods=["DELETE"])
+def deletepoll(poll_id):
+    c = db()
+    conn = c.cursor()
+    variants = conn.execute(
+        f"SELECT text,number FROM variants WHERE id = '{str(poll_id)}'"
+    ).fetchall()
+    
+    if variants == []:
+        return '{"error":"Poll with id %s doesn\'t exist"}' % poll_id, 404
+
+    dict = {}
+    for i in variants:
+        dict[i[0]] = i[1]
+
+    url_admin_token = request.args.get('token')
+    header_admin_token = request.headers.get('Authorization')
+    if url_admin_token:
+        admin_token = url_admin_token
+    elif header_admin_token:
+        t = header_admin_token.split("Bearer ")
+        if t[0] != "":
+            admin_token = t[0]
+        else:
+            admin_token = t[1]
+    else:
+        return '{"error":"Unauthorized"}', 401
+
+    db_admin_token = conn.execute("SELECT admin_token FROM polls WHERE id = '%s'" % str(poll_id)).fetchone()[0]
+    if admin_token != db_admin_token:
+        return '{"error":"Unauthorized"}', 401
+
+    poll = conn.execute("SELECT total,type FROM polls WHERE id = '%s'" % poll_id).fetchone()
+    conn.execute(
+        "DELETE FROM polls WHERE id = '%s'" % poll_id
+    )
+    conn.execute(
+        "DELETE FROM variants WHERE id = '%s'" % poll_id
+    )
+
+    c.commit()
+    c.close()
+    ret = '{"message":"Poll with id %s deleted successfully","results":%s,"total":%s,"type":"%s"}' % (str(poll_id), dict, poll[0], get_poll_type(poll[1]))
+    return ret.replace("\'", "\""), 200
+
+@app.route("/api/v1/stop/<uuid:poll_id>", methods=["POST", "PUT"])
+def stoppoll(poll_id):
+    c = db()
+    conn = c.cursor()
+    variants = conn.execute(
+        f"SELECT text,number FROM variants WHERE id = '{str(poll_id)}'"
+    ).fetchall()
+    
+    if variants == []:
+        return '{"error":"Poll with id %s doesn\'t exist"}' % poll_id, 404
+
+    url_admin_token = request.args.get('token')
+    header_admin_token = request.headers.get('Authorization')
+    if url_admin_token:
+        admin_token = url_admin_token
+    elif header_admin_token:
+        t = header_admin_token.split("Bearer ")
+        if t[0] != "":
+            admin_token = t[0]
+        else:
+            admin_token = t[1]
+    else:
+        return '{"error":"Unauthorized"}', 401
+
+    db_admin_token = conn.execute("SELECT admin_token FROM polls WHERE id = '%s'" % str(poll_id)).fetchone()[0]
+    if admin_token != db_admin_token:
+        return '{"error":"Unauthorized"}', 401
+    
+    type = conn.execute("SELECT type FROM polls WHERE id = '%s'" % poll_id).fetchone()[0]
+    if type == 1:
+        new_type = 2
+    elif type == 3:
+        new_type = 9
+    else:
+        return "", 500
+
+    dict = {}
+    for i in variants:
+        dict[i[0]] = i[1]
+
+    conn.execute(f"""
+        UPDATE polls
+        SET type = '{str(new_type)}'
+        WHERE id = '{str(poll_id)}';
+                 """)
+    
+    poll = conn.execute("SELECT total,type FROM polls WHERE id = '%s'" % poll_id).fetchone()
+    c.commit()
+    c.close()
+
+    ret = '{"message":"Voting %s has ended!","results":%s,"total":%s,"type":"%s"}' % (str(poll_id), dict, poll[0], get_poll_type(poll[1]))
+    return ret.replace("\'", "\""), 200
+
+
+@app.route("/api/v1/polls/<uuid:poll_id>", methods=["GET"])
+def getvariants(poll_id):
+    c = db()
+    conn = c.cursor()
+    variants = conn.execute(
+        f"SELECT text,number FROM variants WHERE id = '{str(poll_id)}'"
+    ).fetchall()
+
+    arr = []
+    dict = {}
+    for i in variants:
+        arr.append(i[0])
+        dict[i[0]] = i[1]
+    if arr == []:
+        return '{"error":"Poll with id %s doesn\'t exist"}' % poll_id, 404
+
+    poll = conn.execute("SELECT total,type FROM polls WHERE id = '%s'" % poll_id).fetchone()
+    
+    c.commit()
+    c.close()
+    ret = '{"variants":%s,"results":%s,"total":%s,"type":"%s"}' % (arr, dict, poll[0], get_poll_type(poll[1]))
+
+    return ret.replace("\'", "\""), 200
+
+@app.route("/api/v1/polls/<uuid:poll_id>", methods=["POST", "PUT"])
+def vote(poll_id):
+    if request.cookies.get(str(poll_id)):
+        return '{"error":"Answer has been already sent by you"}', 429
+
+    conn = db()
+    type = conn.execute("SELECT type FROM polls WHERE id = '%s'" % poll_id).fetchone()[0]
+
+    if type == 1:
+        if not request.args.get("id"):
+            return '{"error":"ID param is unset"}', 400
+        ids = [request.args.get("id")]
+    elif type == 3:
+        ids = []
+        if request.args == {}:
+            return '{"error":"IDs params are unset"}', 400
+        for id, check in request.args.items():
+            if check == "true":
+                ids.append(int(id))
+
+    elif type == 2 or type == 9:
+        return '{"error":"The poll with id %s has ended"}' % str(poll_id), 400
+    else:
+        return "", 500
+    
+    available_variants = conn.execute(
+        f"SELECT text,number FROM variants WHERE id = '{str(poll_id)}'"
+    ).fetchall()
+    available_variants_array = []
+    for i in available_variants:
+        available_variants_array.append(i[0])
+    if available_variants_array == []:
+        return '{"error":"Poll with id %s doesn\'t exist"}' % poll_id, 404
+
+    for varid in ids:
+        varid = int(varid)
+        if varid >= len(available_variants_array):
+            return '{"error:":"Poll %s doesn\'t have variant with id %s"}' % (str(poll_id), str(varid))
+        
+        variant = available_variants[varid]
+        conn.execute(
+            "UPDATE variants SET number = number + 1 WHERE id = '%s' and text = '%s';"
+            % (str(poll_id), variant[0])
+        )
+
+    conn.execute(
+        "UPDATE polls SET total = total + 1 WHERE id = '%s'"
+        % str(poll_id)
+    )
+    poll = conn.execute("SELECT total,type FROM polls WHERE id = '%s'" % poll_id).fetchone()
+
+    dict = {}
+    available_variants = conn.execute(
+        f"SELECT text,number FROM variants WHERE id = '{str(poll_id)}'"
+    ).fetchall()
+    for i in available_variants:
+        dict[i[0]] = i[1]
+
+    conn.commit()
+    conn.close()
+
+    ret = '{"message":"You have successfully voted in %s!","variants":%s,"result":%s,"total":%s,"type":"%s"}' % (str(poll_id), ids, dict, poll[0], get_poll_type(poll[1]))
+    resp = make_response(ret.replace("\'", "\""))
+    resp.set_cookie(str(poll_id), str(ids))
+    return resp, 200
+
+if __name__ == "__main__":
+    app.run()

src/db.py

@@ -0,0 +1,37 @@
+import tomli
+import sqlite3
+
+with open("config.toml", mode="rb") as fp:
+    cfg = tomli.load(fp)
+
+def initdb(conn):
+    c = conn.cursor()
+    if c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='polls';").fetchall() == []:
+
+        print("Creating table 'polls'")
+        c.execute("""
+                  CREATE TABLE polls (
+                    id TEXT PRIMARY KEY NOT NULL,
+                    admin_token TEXT NOT NULL,
+                    title TEXT NOT NULL,
+                    creator TEXT,
+                    description TEXT,
+                    type INTEGER NOT NULL,
+                    alias TEXT,
+                    total INTEGER);
+                  """)
+    if c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='variants';").fetchall() == []:
+
+        print("Creating table 'variants'")
+        c.execute("""
+                  CREATE TABLE variants (
+                    id TEXT NOT NULL,
+                    text TEXT NOT NULL,
+                    number INTEGER NOT NULL);
+                  """)
+    conn.commit()
+    conn.close()
+
+def db():
+    conn = sqlite3.connect(cfg["database"]["file"])
+    return conn