Compare commits

...

14 Commits

4
.gitignore vendored

@ -1,2 +1,6 @@
env/
/dist/
/warren/__pycache__/
/.tox/
*.db
/.coverage

@ -0,0 +1,22 @@
.DEFAULT_GOAL := help
.PHONY: coverage deps help lint push test
coverage: ## Run tests with coverage
coverage erase
coverage run --include=warren/* -m pytest -ra
coverage report -m
deps: ## Install dependencies
pip install black coverage flake8 mccabe mypy pylint pytest tox
lint: ## Lint and static-check
black warren
flake8 warren
pylint warren
mypy warren
push: ## Push code with tags
git push && git push --tags
test: ## Run tests
pytest -ra

@ -16,3 +16,7 @@ Listen for http posts on two routes,
- [ ] gui front end for sql queries
** references
gonna try this new packaging paradigm: https://antonz.org/python-packaging/
** notes to myself
- flit is interesting but REALLY frustrating; if you get something wrong the errors are not very obvious.
- there's some issue with entering a password in the cli as prompted; putting it in the ~.pypirc~ file fixes the issue?

@ -0,0 +1,2 @@
"""Part of the Weir project."""
__version__ = "0.1.0"

@ -1,25 +1,18 @@
#!/usr/bin/env python
"""Part of the Weir project.
This will listen on two endpoints, and when data is
posted to the proper one, matching a particular format,
it will save that data to a local sqlite3 db."""
#!/usr/bin/env python
"""Part of the Weir project."""
__version__ = "0.1.0"
from datetime import datetime
import logging
from datetime import datetime
import os
from pathlib import Path
from flask import Flask, redirect, url_for, request
import pdb
import json
import sqlite3
import pdb
from flask import Flask, request
logging.basicConfig(level=logging.DEBUG)
dbpath = "warren.db"
app = Flask(__name__)
@ -33,8 +26,7 @@ def check_if_db_exists(dbpath):
"""checks to see if the db exists"""
if os.path.exists(dbpath):
return True
else:
return False
return False
def init_new_db(dbpath):
@ -42,61 +34,66 @@ def init_new_db(dbpath):
will automatically create a db file if one is not there.
This function handles explicitly creating the folder structure
and inniting the connection as required.
You can specify your own dbpath, or you can use the default dbpath."""
# TODO: write directory handling shit
# directory handling will go here
# handle the DB stuff
conn = None
try:
conn = sqlite3.connect(dbpath)
cur = conn.cursor()
cur.execute('''CREATE TABLE weir (date, hostname, metadata)''')
cur.execute("""CREATE TABLE weir (date, hostname, metadata)""")
conn.commit()
logging.DEBUG(sqlite3.version)
logging.DEBUG("Creating new weir table.")
logging.debug(sqlite3.version)
logging.debug("Creating new weir table.")
conn.close()
except:
logging.error(logging.ERROR[0])
except sqlite3.OperationalError as error:
logging.error(error)
def write_to_db(data, dbpath):
"""given a python dict and a path for the db, writes dict to db.
calls out to create a new db if one doesn't exist."""
pdb.set_trace()
if check_if_db_exists(dbpath) != True:
db = init_new_db(dbpath)
if check_if_db_exists(dbpath) is not True:
init_new_db(dbpath)
conn = sqlite3.connect(dbpath)
cur = conn.cursor()
right_now = datetime.utcnow().strftime("%Y%m%d")
sql_inputs = [(right_now, data['hostname'], data['metadata'])]
sql_inputs = [(right_now, data["hostname"], data["metadata"])]
cur.executemany("INSERT INTO weir VALUES (?,?,?)", sql_inputs)
conn.commit()
logging.INFO(cur.lastrowid)
# always make sure to close db connection after you're done
conn.close()
@app.route('/')
@app.errorhandler(Exception)
def server_error(err):
"""catch all exception handler"""
logging.exception(err)
return "exception", 400
@app.route("/")
def test():
return 'the website is up you little bitch'
"""healthcheck endpoint with no logic; useful for loadbalancing checks."""
return "the website is up you little bitch"
@app.route('/weir',methods = ['POST'])
@app.route("/weir", methods=["POST"])
def weir():
""""""
try:
python_dict = request.get_json()
write_to_db(python_dict, "warren.db")
return "OK"
except:
return "Error.", 400
"""Primary app route."""
python_dict = request.get_json()
write_to_db(python_dict, "warren.db")
return "OK"
if __name__ == '__main__':
app.run(debug = True)
if __name__ == "__main__":
app.run(debug=True)