ooh, yeah pm me with that.
- app
- __init__
- configs
- routes.py
- config.py
- models
- services
- views
- templates
- trinity
- __init__
- database
- router
- response.py
- request.py
- template
-utilites
- reloader.pyimport pymysql
from DBUtils.PersistentDB import PersistentDB
from trinity import Trinity
from app.configs.config import config
from app.configs.routes import urls
app = Trinity(
routes=urls,
tpl_dir="templates"
)
def connect_db():
return PersistentDB(
creator=pymysql,
host=config['mysql']['host'],
user=config['mysql']['user'],
password=config['mysql']['passw'],
database=config['mysql']['database'],
autocommit=True,
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor
)
def get_db():
if not hasattr(app, 'db'):
app.db = connect_db()
return app.db.connection()app = Flask(
__name__,
template_folder="views/",
static_folder="../pub"
)from . import tpl
from trinity.router import Response
def View(request):
text = "World!"
return Response(tpl.render("home/main.html", test=text))<!DOCTYPE html>
<html>
<head>
<title>HOME</title>
</head>
<body>
Hello {{test}}
</body>
</html>import re
from trinity.template import to_regex
from trinity.router import Request
from app.views import not_found
class Trinity(object):
"""
Main WSGI application. Sets routes and tpl_dir on init
and dispatches Request object on __call__. Dispatcher
then matches _req_url with _urls set by Trinity.
"""
def __init__(self, routes=dict(), tpl_dir=""):
self._routes = routes
self._tpl_dir = tpl_dir
def __call__(self, environ, start_response):
response = self.dispatch(Request(environ))
start_response(response.status, response.wsgi_headers)
# Yield dispatcher response
yield response.body.encode('UTF-8')
def dispatch(self, request):
_req_url = request.path
for _url, View in self._routes.items():
match = re.search(to_regex(_url), _req_url)
if match is not None:
# Send request params and return View
request.params = match.groups()
return View(request)
# If match is None
return not_found.View(request)from urllib.parse import parse_qs
class Request(object):
"""
Creates a request object.
"""
def __init__(self, environ):
self.environ = environ
self.path = environ["PATH_INFO"]
self.method = environ['REQUEST_METHOD']
self.headers = {k: v for k, v in environ.items() if k.startswith("HTTP_")}
self.params = None
self.GET = parse_qs(environ["QUERY_STRING"])
self.POST = {}
# Experimental
if environ.get('HTTPS', 'off') in ('on', '1'):
self.environ['wsgi.url_scheme'] = 'https'
else:
self.environ['wsgi.url_scheme'] = 'http'
if self.method == "POST":
try:
content_length = int(environ.get("CONTENT_LENGTH", 0))
except ValueError:
content_length = 0
query_string = environ["wsgi.input"].read(content_length)
self.POST = parse_qs(query_string)class Response(object):
"""
Creates response object, binding headers, status and body.
"""
def __init__(self, body, status="200 OK"):
self.status = status
self.body = str(body) or ""
self.headers = {
"Content-Type": "text/html",
"Content-Length": str(len(self.body)),
"X-Powered-By": "Trinity-py-tr4.5"
}
@property
def wsgi_headers(self):
return [(k, v) for k, v in self.headers.items()]import re
from jinja2 import Environment, PackageLoader
class Jinja2:
"""
Jinja2 templating wrapper
"""
def __init__(self, tpls_dir):
self.jinja2_env = Environment(
loader=PackageLoader("app", tpls_dir),
autoescape=True)
def render(self, name, **ctx):
t = self.jinja2_env.get_template(name)
return t.render(ctx)
def to_regex(template):
"""
Converts string to templated regex
"""
regex = ''
var_regex = re.compile(r'''\{(\w+)(?::([^}]+))? \}''', re.VERBOSE)
last_pos = 0
for match in var_regex.finditer(template):
regex += re.escape(template[last_pos:match.start()])
var_name = match.group(1)
expr = match.group(2) or '[^/]+'
expr = '(?P<%s>%s)' % (var_name, expr)
regex += expr
last_pos = match.end()
regex += re.escape(template[last_pos:])
regex = '^%s$' % regex
return regex
from datetime import datetime
from sys import version_info
from wsgiref.simple_server import make_server
from app import app
from app.configs.config import config
class TrinityDevServer:
""" Trinity development server """
def __init__(self):
print("[DEV][INFO] Note: Not intended for production use, refer to readme.md for more info.")
self.show_splash()
self.load_config()
self.create_server()
def show_splash(self):
_pyver = ".".join(map(str, version_info[:3]))
_splash = '''
_______ _ _
(_______) (_) (_) _
_ ____ _ ____ _ _| |_ _ _ _____ ____ _ _
| |/ ___) | _ \\| (_ _) | | (_____) _ \\| | | |
| | | | | | | | | | |_| |_| | | |_| | |_| |
|_|_| |_|_| |_|_| \\__)\\__ | | __/ \\__ |
(____/ |_| (____/
'''
print('{:^60}'.format(_splash))
print(f"[DEV][ OK ] Python {_pyver}\n")
def load_config(self):
_error = False
print("[DEV][INFO] Requesting config ..")
try:
self.config = config
except Exception as error:
_error = True
print("\n[DEV][ERRO] "+str(error))
exit()
if _error != True: print("[DEV][ OK ] Configuration loaded! ")
def create_server(self):
_error = False
_parsed = str(self.config['trinity']['ip'])+\
":"+str(self.config['trinity']['port'])
print(f"\n[DEV][INFO] Binding to {_parsed}..")
try:
with make_server(
self.config['trinity']['ip'],
self.config['trinity']['port'], app) as httpd:
if _error != True:
_date = datetime.now().strftime('%H:%M%p - %m-%d-%Y')
print(f"[DEV][ OK ] READY! ({_parsed}) ({_date})\n")
httpd.serve_forever()
except Exception as error:
_error = True
print("\n[DEV][ERRO] "+str(error))
exit()
if __name__ == '__main__':
TrinityDevServer()import sys
import os
import time
import threading
import subprocess
import tkinter.ttk as ttk
from tkinter import *
from tkinter import messagebox
from dev_server import TrinityDevServer
class GUI:
def __init__(self, master, arg=""):
_width = 650
_height = 460
self.active = None
self.master = master
_title = master.title("Trinity Dev Server GUI")
master.geometry(("%dx%d")%(_width, _height))
master.maxsize(1000, 500)
style = ttk
style.Style().theme_use("clam")
self.start_button = style.Button(master,
text="Start server",
command=self.load_server_instance
)
self.start_button.place(x = 10,y = 15)
self.reload_button = style.Button(master,
text="Force reload",
command=self.reload_server_instance
)
self.reload_button.place(x = 120,y = 15)
#areaOutput
self.areaOutput = Text(master, fg="orange", bg="black")
self.areaOutput.pack(expand=1, side='left')
self.areaOutput.place(y=60)
sys.stdout = self
if arg == 'restart':
print("[DEV][ OK ] Server restarted ! \n")
self.load_server_instance()
else:
print(arg)
def load_server_instance(self):
if self.active != True:
self.active = True
def callback():
self.t = threading.Thread(target=TrinityDevServer)
self.t.daemon = True
self.t.start()
self.master.after_idle(callback)
else:
print("[DEV][ERRO] Server instance already active")
def reload_server_instance(self):
def callback():
if self.active == True:
subprocess.Popen(
[sys.executable, os.path.abspath(__file__), 'restart'])
time.sleep(0.200)
exit()
else:
print("[DEV][INFO] No server instance to reload.")
self.master.after_idle(callback)
def write(self, txt):
self.areaOutput.insert(END, str(txt))
root = Tk()
try:
arg = sys.argv[1]
except:
arg = '''
Welcome to the Trinity Dev Server GUI.\n
Click `Start server` above to start serving.\n
Click `Force reload` to reload the server instance
'''
gui = GUI(root, arg)
root.mainloop()""" Wsgi for passenger phusion """
from app import app as application[/B][/SIZE]
$ gunicorn app:app -w 4 -b 0.0.0.0:8000waitress-serve --listen=127.0.0.1:8000 app:app
[SIZE=4][B]from wsgiref.handlers import CGIHandler
from app import app
CGIHandler().run(app)fastcgi_wsgi.fcgi
from flup.server.fcgi import WSGIServer
from app import app as application
WSGIServer(application).run()
