1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111
|
from os import path
from random import randrange
from random import sample
from jinja2 import Environment
from jinja2 import FileSystemLoader
from sqlalchemy import MetaData
from sqlalchemy.orm import create_session
from sqlalchemy.orm import scoped_session
from werkzeug.local import Local
from werkzeug.local import LocalManager
from werkzeug.routing import Map
from werkzeug.routing import Rule
from werkzeug.urls import url_parse
from werkzeug.utils import cached_property
from werkzeug.wrappers import Response
TEMPLATE_PATH = path.join(path.dirname(__file__), "templates")
STATIC_PATH = path.join(path.dirname(__file__), "static")
ALLOWED_SCHEMES = frozenset(["http", "https", "ftp", "ftps"])
URL_CHARS = "abcdefghijkmpqrstuvwxyzABCDEFGHIJKLMNPQRST23456789"
local = Local()
local_manager = LocalManager([local])
application = local("application")
metadata = MetaData()
url_map = Map([Rule("/static/<file>", endpoint="static", build_only=True)])
session = scoped_session(
lambda: create_session(
application.database_engine, autocommit=False, autoflush=False
)
)
jinja_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH))
def expose(rule, **kw):
def decorate(f):
kw["endpoint"] = f.__name__
url_map.add(Rule(rule, **kw))
return f
return decorate
def url_for(endpoint, _external=False, **values):
return local.url_adapter.build(endpoint, values, force_external=_external)
jinja_env.globals["url_for"] = url_for
def render_template(template, **context):
return Response(
jinja_env.get_template(template).render(**context), mimetype="text/html"
)
def validate_url(url):
return url_parse(url)[0] in ALLOWED_SCHEMES
def get_random_uid():
return "".join(sample(URL_CHARS, randrange(3, 9)))
class Pagination(object):
def __init__(self, query, per_page, page, endpoint):
self.query = query
self.per_page = per_page
self.page = page
self.endpoint = endpoint
@cached_property
def count(self):
return self.query.count()
@cached_property
def entries(self):
return (
self.query.offset((self.page - 1) * self.per_page)
.limit(self.per_page)
.all()
)
@property
def has_previous(self):
"""Return True if there are pages before the current one."""
return self.page > 1
@property
def has_next(self):
"""Return True if there are pages after the current one."""
return self.page < self.pages
@property
def previous(self):
"""Return the URL for the previous page."""
return url_for(self.endpoint, page=self.page - 1)
@property
def next(self):
"""Return the URL for the next page."""
return url_for(self.endpoint, page=self.page + 1)
@property
def pages(self):
"""Return the number of pages."""
return max(0, self.count - 1) // self.per_page + 1
|