Skip to content

Commit

Permalink
introduce Configuration and ConfigurationProvider
Browse files Browse the repository at this point in the history
The Configuration class will help structure the configuration better and will allow us to mock values easier when testing.

The ConfigurationProvider is responsible for providing the same config across the entire app through late binding.
  • Loading branch information
shtlrs authored and supakeen committed Mar 2, 2024
1 parent 295e43a commit 049421a
Show file tree
Hide file tree
Showing 16 changed files with 219 additions and 95 deletions.
7 changes: 6 additions & 1 deletion src/pinnwand/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,17 @@

import tornado.web

from pinnwand import configuration, handler, logger, path
from pinnwand import handler, logger, path
from pinnwand.configuration import Configuration, ConfigurationProvider

log = logger.get_logger(__name__)


def make_application(debug: bool = False) -> tornado.web.Application:
configuration: Configuration = ConfigurationProvider.get_config(
load_env=True
)

pages: List[Any] = [
(r"/", handler.website.Create),
(r"/\+(.*)", handler.website.Create),
Expand Down
47 changes: 9 additions & 38 deletions src/pinnwand/command.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,17 +2,16 @@
a HTTP server, add and remove paste, initialize the database and reap expired
pastes."""

import ast
import logging
import os
import sys
from datetime import timedelta
from typing import TYPE_CHECKING, Optional
from typing import Optional

import click
import tornado.ioloop

from pinnwand import logger
from pinnwand.configuration import Configuration, ConfigurationProvider
from pinnwand.database import models, manager, utils

log = logger.get_logger(__name__)
Expand All @@ -30,42 +29,12 @@ def main(verbose: int, configuration_path: Optional[str]) -> None:
"""Pinnwand pastebin software."""
logging.basicConfig(level=10 + (logging.FATAL - verbose * 10))

from pinnwand import configuration

configuration: Configuration = ConfigurationProvider.get_config()
# First check if we have a configuration path
if configuration_path:
if (
TYPE_CHECKING
): # lie to mypy, see https://github.com/python/mypy/issues/1153
import tomllib as toml
else:
try:
import tomllib as toml
except ImportError:
import tomli as toml

with open(configuration_path, "rb") as file:
configuration_file = toml.load(file)

for key, value in configuration_file.items():
setattr(configuration, key, value)

# Or perhaps we have configuration in the environment, these are prefixed
# with PINNWAND_ and all upper case, remove the prefix, convert to
# lowercase.
for key, value in os.environ.items():
if key.startswith("PINNWAND_"):
key = key.removeprefix("PINNWAND_")
key = key.lower()

try:
value = ast.literal_eval(value)
except (ValueError, SyntaxError):
# When `ast.literal_eval` can't parse the value into another
# type we take it at string value
pass

setattr(configuration, key, value)
configuration.load_config_file(configuration_path)

configuration.load_environment()

engine = manager.DatabaseManager.get_engine()
utils.create_tables(engine)
Expand All @@ -81,9 +50,11 @@ def main(verbose: int, configuration_path: Optional[str]) -> None:
)
def http(port: int, debug: bool) -> None:
"""Run pinnwand's HTTP server."""
from pinnwand import configuration, utility
from pinnwand import utility
from pinnwand.app import make_application

configuration: Configuration = ConfigurationProvider.get_config()

# Reap expired pastes on startup (we might've been shut down for a while)
utility.reap()

Expand Down
176 changes: 146 additions & 30 deletions src/pinnwand/configuration.py
Original file line number Diff line number Diff line change
@@ -1,30 +1,146 @@
database_uri = "sqlite:///:memory:"
paste_size = 256 * 1024 # in bytes
footer = 'View <a href="//github.com/supakeen/pinnwand" target="_BLANK">source code</a>, the <a href="/removal">removal</a> or <a href="/expiry">expiry</a> stories, or read the <a href="/about">about</a> page.'
paste_help = "<p>Welcome to pinnwand, this site is a pastebin. It allows you to share code with others. If you write code in the text area below and press the paste button you will be given a link you can share with others so they can view your code as well.</p><p>People with the link can view your pasted code, only you can remove your paste and it expires automatically. Note that anyone could guess the URI to your paste so don't rely on it being private.</p>"
page_path = None
page_list = ["about", "removal", "expiry"]
default_selected_lexer = "text"
preferred_lexers = [] # type: ignore
logo_path = None
report_email = None
expiries = {"1day": 86400, "1week": 604800}
reaping_periodicity = 1_800_800
ratelimit = {
"read": {
"capacity": 100,
"consume": 1,
"refill": 2,
},
"create": {
"capacity": 2,
"consume": 2,
"refill": 1,
},
"delete": {
"capacity": 2,
"consume": 2,
"refill": 1,
},
}
spamscore = 50
import ast
import os
from typing import Optional, TYPE_CHECKING

if TYPE_CHECKING: # lie to mypy, see https://github.com/python/mypy/issues/1153
import tomllib as toml
else:
try:
import tomllib as toml
except ImportError:
import tomli as toml


class Configuration:
"""A class that holds all pinnwand's configuration."""

def __init__(self):
self._database_uri = "sqlite:///:memory:"
self._paste_size = 256 * 1024 # in bytes
self._footer = 'View <a href="//github.com/supakeen/pinnwand" target="_BLANK">source code</a>, the <a href="/removal">removal</a> or <a href="/expiry">expiry</a> stories, or read the <a href="/about">about</a> page.'
self._paste_help = "<p>Welcome to pinnwand, this site is a pastebin. It allows you to share code with others. If you write code in the text area below and press the paste button you will be given a link you can share with others so they can view your code as well.</p><p>People with the link can view your pasted code, only you can remove your paste and it expires automatically. Note that anyone could guess the URI to your paste so don't rely on it being private.</p>"
self._page_path = None
self._page_list = ["about", "removal", "expiry"]
self._default_selected_lexer = "text"
self._preferred_lexers = [] # type: ignore
self._logo_path = None
self._report_email = None
self._expiries = {"1day": 86400, "1week": 604800}
self._reaping_periodicity = 1_800_800
self._ratelimit = {
"read": {
"capacity": 100,
"consume": 1,
"refill": 2,
},
"create": {
"capacity": 2,
"consume": 2,
"refill": 1,
},
"delete": {
"capacity": 2,
"consume": 2,
"refill": 1,
},
}
self._spamscore = 50

# Define getters for each configuration parameter
@property
def database_uri(self):
return self._database_uri

@property
def paste_size(self):
return self._paste_size

@property
def footer(self):
return self._footer

@property
def paste_help(self):
return self._paste_help

@property
def page_path(self):
return self._page_path

@property
def page_list(self):
return self._page_list

@property
def default_selected_lexer(self):
return self._default_selected_lexer

@property
def preferred_lexers(self):
return self._preferred_lexers

@property
def logo_path(self):
return self._logo_path

@property
def report_email(self):
return self._report_email

@property
def expiries(self):
return self._expiries

@property
def reaping_periodicity(self):
return self._reaping_periodicity

@property
def ratelimit(self):
return self._ratelimit

@property
def spamscore(self):
return self._spamscore

def load_config_file(self, path: Optional[str] = None) -> None:
"""Load configuration settings from a toml file."""

with open(path, "rb") as file:
loaded_configuration = toml.load(file)

for key, value in loaded_configuration.items():
setattr(self, f"_{key}", value)

def load_environment(self):
"""Load configuration from the environment, if any."""
for key, value in os.environ.items():
if not key.startswith("PINNWAND_"):
continue

key = key.removeprefix("PINNWAND_")
key = key.lower()

try:
value = ast.literal_eval(value)
setattr(self, f"_{key}", value)
except (ValueError, SyntaxError):
# When `ast.literal_eval` can't parse the value into another
# type we take it at string value
pass


class ConfigurationProvider:
"""A class responsible for providing an instance of the configuration on demand."""

_config: Configuration = None

@classmethod
def get_config(cls, load_env=False):
"""Return the loaded configuration."""
if not cls._config:
cls._config = Configuration()
if load_env:
cls._config.load_environment()

return cls._config
4 changes: 2 additions & 2 deletions src/pinnwand/database/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
from sqlalchemy.orm.session import Session


from pinnwand import configuration
from pinnwand.configuration import Configuration, ConfigurationProvider


class DatabaseManager:
Expand All @@ -17,7 +17,7 @@ class DatabaseManager:
@classmethod
def get_engine(cls):
"""Return an engine for the currently configured connection string."""

configuration: Configuration = ConfigurationProvider.get_config()
if not cls._engine:
cls._engine = create_engine(configuration.database_uri)

Expand Down
5 changes: 4 additions & 1 deletion src/pinnwand/database/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,8 @@
relationship,
)

from pinnwand import configuration, defensive, error, logger, utility
from pinnwand import defensive, error, logger, utility
from pinnwand.configuration import Configuration, ConfigurationProvider


log = logger.get_logger(__name__)
Expand Down Expand Up @@ -102,6 +103,8 @@ def __init__(
if not len(raw):
raise error.ValidationError("Empty pastes are not allowed")

configuration: Configuration = ConfigurationProvider.get_config()

if len(raw) > configuration.paste_size:
raise error.ValidationError(
f"Text exceeds size limit {configuration.paste_size//1024} (kB)"
Expand Down
5 changes: 4 additions & 1 deletion src/pinnwand/defensive.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,8 @@
import token_bucket
from tornado.httputil import HTTPServerRequest

from pinnwand import configuration, logger
from pinnwand import logger
from pinnwand.configuration import Configuration, ConfigurationProvider

log = logger.get_logger(__name__)

Expand All @@ -30,6 +31,8 @@ def ratelimit(request: HTTPServerRequest, area: str = "global") -> bool:
ranges usually belong to the same person. If this is encountered in real
life this function will have to be expanded."""

configuration: Configuration = ConfigurationProvider.get_config()

if area not in ratelimit_area:
ratelimit_area[area] = {}

Expand Down
4 changes: 3 additions & 1 deletion src/pinnwand/handler/api_curl.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,8 @@

import tornado.web

from pinnwand import configuration, defensive, logger, utility
from pinnwand import defensive, logger, utility
from pinnwand.configuration import Configuration, ConfigurationProvider
from pinnwand.database import models, manager

log = logger.get_logger(__name__)
Expand All @@ -21,6 +22,7 @@ def post(self) -> None:
self.write("Enhance your calm, you have exceeded the ratelimit.")
return

configuration: Configuration = ConfigurationProvider.get_config()
lexer = self.get_body_argument("lexer", "text")
raw = self.get_body_argument("raw", "", strip=False)
expiry = self.get_body_argument("expiry", "1day")
Expand Down
7 changes: 6 additions & 1 deletion src/pinnwand/handler/api_deprecated.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,8 @@
import tornado.web
from tornado.escape import url_escape

from pinnwand import configuration, defensive, error, logger, utility
from pinnwand import defensive, error, logger, utility
from pinnwand.configuration import Configuration, ConfigurationProvider
from pinnwand.database import models, manager

log = logger.get_logger(__name__)
Expand Down Expand Up @@ -124,6 +125,8 @@ async def post(self) -> None:
if defensive.ratelimit(self.request, area="create"):
raise error.RatelimitError()

configuration: Configuration = ConfigurationProvider.get_config()

lexer = self.get_body_argument("lexer")
raw = self.get_body_argument("code", strip=False)
expiry = self.get_body_argument("expiry")
Expand Down Expand Up @@ -222,6 +225,8 @@ async def get(self) -> None:
if defensive.ratelimit(self.request, area="read"):
raise error.RatelimitError()

configuration: Configuration = ConfigurationProvider.get_config()

self.write(
{
name: str(timedelta(seconds=delta))
Expand Down
Loading

0 comments on commit 049421a

Please sign in to comment.