Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

web: Add types #371

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion elixir/filters/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ class FilterContext:
tag: str
family: str
filepath: str
get_ident_url: str
get_ident_url: Callable[[str], str]
get_absolute_source_url: Callable[[str], str]
get_relative_source_url: Callable[[str], str]

Expand Down
84 changes: 37 additions & 47 deletions elixir/web.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import datetime
from collections import OrderedDict, namedtuple
from re import search, sub
from typing import Any, Callable, NamedTuple, Tuple
from urllib import parse
import falcon
import jinja2
Expand All @@ -39,7 +40,7 @@
from .api import ApiIdentGetterResource
from .query import get_query
from .web_utils import ProjectConverter, IdentConverter, validate_version, validate_project, validate_ident, \
get_elixir_version_string, get_elixir_repo_link
get_elixir_version_string, get_elixir_repo_link, RequestContext, Config

VERSION_CACHE_DURATION_SECONDS = 2 * 60 # 2 minutes
ADD_ISSUE_LINK = "https://github.com/bootlin/elixir/issues/new"
Expand Down Expand Up @@ -187,12 +188,12 @@ def validate_project_and_version(ctx, project, version):


# Returns base url of source pages
# project and version assumed unquoted
def get_source_base_url(project, version):
# project and version are assumed to be unquoted
def get_source_base_url(project: str, version: str) -> str:
return f'/{ parse.quote(project, safe="") }/{ parse.quote(version, safe="") }/source'

# Converts ParsedSourcePath to a string with corresponding URL path
def stringify_source_path(project, version, path):
def stringify_source_path(project: str, version: str, path: str) -> str:
if not path.startswith('/'):
path = '/' + path
path = f'{ get_source_base_url(project, version) }{ path }'
Expand All @@ -217,7 +218,7 @@ def on_get(self, req, resp):
# Handles source URLs
# Path parameters are asssumed to be unquoted by converters
class SourceResource:
def on_get(self, req, resp, project, version, path):
def on_get(self, req, resp, project: str, version: str, path: str):
project, version, query = validate_project_and_version(req.context, project, version)

if not path.startswith('/') and len(path) != 0:
Expand Down Expand Up @@ -251,13 +252,13 @@ def on_get(self, req, resp, project, version, path):
# Handles source URLs without a path, ex. '/u-boot/v2023.10/source'.
# Note lack of trailing slash
class SourceWithoutPathResource(SourceResource):
def on_get(self, req, resp, project, version):
def on_get(self, req, resp, project: str, version: str):
return super().on_get(req, resp, project, version, '')


# Returns base url of ident pages
# project and version assumed unquoted
def get_ident_base_url(project, version, family=None):
def get_ident_base_url(project: str, version: str, family: str|None = None) -> str:
project = parse.quote(project, safe="")
version = parse.quote(version, safe="")
if family is not None:
Expand All @@ -266,7 +267,7 @@ def get_ident_base_url(project, version, family=None):
return f'/{ project }/{ version }/ident'

# Converts ParsedIdentPath to a string with corresponding URL path
def stringify_ident_path(project, version, family, ident):
def stringify_ident_path(project, version, family, ident) -> str:
path = f'{ get_ident_base_url(project, version, family) }/{ parse.quote(ident, safe="") }'
return path.rstrip('/')

Expand All @@ -277,7 +278,7 @@ def on_get(self, req, resp, project, version, family=None, ident=None):
resp.status = falcon.HTTP_FOUND
resp.location = stringify_source_path(project, version, "")

def on_post(self, req, resp, project, version, family=None, ident=None):
def on_post(self, req, resp, project: str, version: str, family: str|None = None, _ident: str|None = None):
project, version, query = validate_project_and_version(req.context, project, version)

form = req.get_media()
Expand Down Expand Up @@ -305,7 +306,7 @@ def on_post(self, req, resp, project, version, family=None, ident=None):
# See IdentPostRedirectResource for behavior on POST
# Path parameters are asssumed to be unquoted by converters
class IdentResource(IdentPostRedirectResource):
def on_get(self, req, resp, project, version, family, ident):
def on_get(self, req, resp, project: str, version: str, family: str, ident: str):
project, version, query = validate_project_and_version(req.context, project, version)

family = parse.unquote(family)
Expand Down Expand Up @@ -339,7 +340,7 @@ def on_get(self, req, resp, project, version, family, ident):
# Also handles POST requests for ident URLs without family - IdentPostRedirectResource is
# inherited from IdentResource
class IdentWithoutFamilyResource(IdentResource):
def on_get(self, req, resp, project, version, ident):
def on_get(self, req, resp, project: str, version: str, ident: str):
super().on_get(req, resp, project, version, 'C', ident)


Expand All @@ -353,7 +354,7 @@ def on_get(self, req, resp, project, version, ident):
}

# Returns a list of names of top-level directories in basedir
def get_directories(basedir):
def get_directories(basedir: str) -> list[str]:
directories = []
for filename in os.listdir(basedir):
filepath = os.path.join(basedir, filename)
Expand All @@ -366,7 +367,7 @@ def get_directories(basedir):
ProjectEntry = namedtuple('ProjectEntry', 'name, url')

# Returns a list of ProjectEntry tuples of projects stored in directory basedir
def get_projects(basedir):
def get_projects(basedir: str) -> list[ProjectEntry]:
return [ProjectEntry(p, f"/{p}/latest/source") for p in get_directories(basedir)]

# Tuple of version name and URL to chosen resource with that version
Expand All @@ -382,7 +383,10 @@ def get_projects(basedir):
# get_url: function that takes a version string and returns the URL
# for that version. Meaning of the URL can depend on the context
# current_version: string with currently browsed version
def get_versions(versions, get_url, current_version):
def get_versions(versions: OrderedDict[str, OrderedDict[str, str]],
get_url: Callable[[str], str],
current_version: str) -> Tuple[dict[str, dict[str, list[VersionEntry]]], Tuple[str|None, str|None, str|None]]:

result = OrderedDict()
current_version_path = (None, None, None)
for major, minor_verions in versions.items():
Expand Down Expand Up @@ -413,12 +417,11 @@ def get_versions_cached(q, ctx, project):
return cached_versions[1]

# Retruns template context used by the layout template
# q: Query object
# ctx: RequestContext object
# get_url_with_new_version: see get_url parameter of get_versions
# project: name of the project
# version: version of the project
def get_layout_template_context(q, ctx, get_url_with_new_version, project, version):
def get_layout_template_context(q: Query, ctx: RequestContext, get_url_with_new_version: Callable[[str], str],
project: str, version: str) -> dict[str, Any]:
versions_raw = get_versions_cached(q, ctx, project)
versions, current_version_path = get_versions(versions_raw, get_url_with_new_version, version)

Expand Down Expand Up @@ -454,7 +457,7 @@ def generate_raw_source(resp, query, project, version, path):
resp.headers['Content-Security-Policy'] = "sandbox; default-src 'none'"

# Guesses file format based on filename, returns code formatted as HTML
def format_code(filename, code):
def format_code(filename: str, code: str) -> str:
import pygments
import pygments.lexers
import pygments.formatters
Expand Down Expand Up @@ -484,7 +487,7 @@ def format_code(filename, code):
# project: name of the requested project
# version: requested version of the project
# path: path to the file in the repository
def generate_source(q, project, version, path):
def generate_source(q: Query, project: str, version: str, path: str) -> str:
code = q.query('file', version, path)

_, fname = os.path.split(path)
Expand Down Expand Up @@ -526,19 +529,18 @@ def get_ident_url(ident, ident_family=None):
return html_code_block

# Represents a file entry in git tree
# type: either tree (directory), blob (file) or symlink
# type : either tree (directory), blob (file) or symlink
# name: filename of the file
# path: path of the file, path to the target in case of symlinks
# url: absolute URL of the file
# size: int, file size in bytes, None for directories and symlinks
DirectoryEntry = namedtuple('DirectoryEntry', 'type, name, path, url, size')

# Returns a list of DirectoryEntry objects with information about files in a directory
# q: Query object
# base_url: file URLs will be created by appending file path to this URL. It shouldn't end with a slash
# tag: requested repository tag
# path: path to the directory in the repository
def get_directory_entries(q, base_url, tag, path):
def get_directory_entries(q: Query, base_url, tag: str, path: str) -> list[DirectoryEntry]:
dir_entries = []
lines = q.query('dir', tag, path)

Expand All @@ -547,27 +549,25 @@ def get_directory_entries(q, base_url, tag, path):
file_path = f"{ path }/{ name }"

if type == 'tree':
dir_entries.append(('tree', name, file_path, f"{ base_url }{ file_path }", None))
dir_entries.append(DirectoryEntry('tree', name, file_path, f"{ base_url }{ file_path }", None))
elif type == 'blob':
# 120000 permission means it's a symlink
if perm == '120000':
dir_path = path if path.endswith('/') else path + '/'
link_contents = q.get_file_raw(tag, file_path)
link_target_path = os.path.abspath(dir_path + link_contents)

dir_entries.append(('symlink', name, link_target_path, f"{ base_url }{ link_target_path }", size))
dir_entries.append(DirectoryEntry('symlink', name, link_target_path, f"{ base_url }{ link_target_path }", size))
else:
dir_entries.append(('blob', name, file_path, f"{ base_url }{ file_path }", size))
dir_entries.append(DirectoryEntry('blob', name, file_path, f"{ base_url }{ file_path }", size))

return dir_entries

# Generates response (status code and optionally HTML) of the `source` route
# ctx: RequestContext
# q: Query object
# parsed_path: ParsedSourcePath
def generate_source_page(ctx, q, project, version, path):
status = falcon.HTTP_OK
def generate_source_page(ctx: RequestContext, q: Query,
project: str, version: str, path: str) -> tuple[int, str]:

status = falcon.HTTP_OK
source_base_url = get_source_base_url(project, version)

type = q.query('type', version, path)
Expand Down Expand Up @@ -632,14 +632,14 @@ def generate_source_page(ctx, q, project, version, path):
LineWithURL = namedtuple('LineWithURL', 'lineno, url')

# Represents a symbol occurrence to be rendered by ident template
# type: type of the symbol
# type : type of the symbol
# path: path of the file that contains the symbol
# line: list of LineWithURL
SymbolEntry = namedtuple('SymbolEntry', 'type, path, lines')

# Converts SymbolInstance into SymbolEntry
# path of SymbolInstance will be appended to base_url
def symbol_instance_to_entry(base_url, symbol):
def symbol_instance_to_entry(base_url: str, symbol: SymbolInstance) -> SymbolEntry:
# TODO this should be a responsibility of Query
if type(symbol.line) is str:
line_numbers = symbol.line.split(',')
Expand All @@ -654,16 +654,13 @@ def symbol_instance_to_entry(base_url, symbol):
return SymbolEntry(symbol.type, symbol.path, lines)

# Generates response (status code and optionally HTML) of the `ident` route
# ctx: RequestContext
# basedir: path to data directory, ex: "/srv/elixir-data"
# parsed_path: ParsedIdentPath
def generate_ident_page(ctx, q, project, version, family, ident):
status = falcon.HTTP_OK
def generate_ident_page(ctx: RequestContext, q: Query,
project: str, version: str, family: str, ident: str) -> tuple[int, str]:

status = falcon.HTTP_OK
source_base_url = get_source_base_url(project, version)

symbol_definitions, symbol_references, symbol_doccomments = q.query('ident', version, ident, family)

symbol_sections = []

if len(symbol_definitions) or len(symbol_references):
Expand Down Expand Up @@ -721,13 +718,6 @@ def generate_ident_page(ctx, q, project, version, family, ident):
return (status, template.render(data))


# Elixir config, currently contains only path to directory with projects
Config = namedtuple('Config', 'project_dir, version_string, repo_link')

# Basic information about handled request - current Elixir configuration, configured Jinja environment
# and logger
RequestContext = namedtuple('RequestContext', 'config, jinja_env, logger, versions_cache, versions_cache_lock')

def get_jinja_env():
script_dir = os.path.dirname(os.path.realpath(__file__))
templates_dir = os.path.join(script_dir, '../templates/')
Expand All @@ -738,7 +728,7 @@ def get_jinja_env():
# Replaces the default, unquoted URL with a quoted version
# NOTE: this is non-standard and it's not guaranteed to work on all WSGI servers
class RawPathComponent:
def process_request(self, req, resp):
def process_request(self, req, _):
raw_uri = req.env.get('RAW_URI') or req.env.get('REQUEST_URI')
if raw_uri:
req.path, _, _ = raw_uri.partition('?')
Expand All @@ -750,7 +740,7 @@ def __init__(self, jinja_env):
self.versions_cache = {}
self.versions_cache_lock = threading.Lock()

def process_request(self, req, resp):
def process_request(self, req, _resp):
req.context = RequestContext(
Config(req.env['LXR_PROJ_DIR'], ELIXIR_VERSION_STRING, ELIXIR_REPO_LINK),
self.jinja_env,
Expand Down
30 changes: 24 additions & 6 deletions elixir/web_utils.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,11 @@
import os
import re
from urllib import parse
import logging
import threading
from urllib import parse
from typing import Any, Dict, NamedTuple
import falcon
import jinja2

from .lib import validFamily, run_cmd

Expand Down Expand Up @@ -36,30 +39,45 @@ def get_elixir_repo_link(version):
else:
return ELIXIR_REPO_LINK

def validate_project(project):
# Elixir config, currently contains only path to directory with projects
class Config(NamedTuple):
project_dir: str
version_string: str
repo_link: str

# Basic information about handled request - current Elixir configuration, configured Jinja environment
# and logger
class RequestContext(NamedTuple):
config: Config
jinja_env: jinja2.Environment
logger: logging.Logger
versions_cache: Dict[str, str]
versions_cache_lock: threading.Lock

def validate_project(project: str) -> str|None:
if project is not None and re.match(r'^[a-zA-Z0-9_.,:/-]+$', project):
return project.strip()

# Validates and unquotes project parameter
class ProjectConverter(falcon.routing.BaseConverter):
def convert(self, value: str):
def convert(self, value: str) -> str:
value = parse.unquote(value)
project = validate_project(value)
if project is None:
raise falcon.HTTPBadRequest('Error', 'Invalid project name')
return project

def validate_version(version):
def validate_version(version) -> str|None:
if version is not None and re.match(r'^[a-zA-Z0-9_.,:/-]+$', version):
return version.strip()

def validate_ident(ident):
def validate_ident(ident: str) -> str|None:
if ident is not None and re.match(r'^[A-Za-z0-9_,.+?#-]+$', ident):
return ident.strip()

# Validates and unquotes identifier parameter
class IdentConverter(falcon.routing.BaseConverter):
def convert(self, value: str):
def convert(self, value: str) -> str|None:
value = parse.unquote(value)
return validate_ident(value)