You're Invited:Meet the Socket Team at RSAC and BSidesSF 2026, March 23–26.RSVP →
Socket
Book a DemoSign in
Socket
Book a DemoSign in

@vercel/python

Package Overview
Dependencies
Maintainers
2
Versions
230
Alerts
File Explorer

Advanced tools

License
Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

@vercel/python - npm Package Compare versions

Comparing version
6.20.2
 to
6.21.0
+36
templates/vc_django_settings.py
"""
Discover Django settings by running manage.py with a patched
execute_from_command_line. This lets manage.py set DJANGO_SETTINGS_MODULE
via whatever mechanism it uses (os.environ.setdefault, conditionals, etc.),
then loads the settings module and prints all uppercase attributes as JSON.
"""
import os, sys, json, runpy
import django.core.management
class _Escape(Exception):
pass
def _stop(*a, **kw):
raise _Escape
django.core.management.execute_from_command_line = _stop
sys.argv = ["manage.py"]
try:
runpy.run_path("manage.py", run_name="__main__")
except _Escape:
pass
settings_module = os.environ.get("DJANGO_SETTINGS_MODULE")
if not settings_module:
print(json.dumps(None))
else:
import importlib
mod = importlib.import_module(settings_module)
settings = {k: getattr(mod, k) for k in dir(mod) if k.isupper()}
print(json.dumps(settings, default=str))
+596
templates/vc_init_dev.py
# Auto-generated template used by vercel dev (Python, ASGI/WSGI)
# Serves static files from PUBLIC_DIR before delegating to the user app.
import sys
import os
import inspect
import logging
import logging.config
from os import path as _p
from importlib import util as _importlib_util
import mimetypes
# Simple ANSI coloring. Respect NO_COLOR environment variable.
_NO_COLOR = "NO_COLOR" in os.environ
_RESET = "\x1b[0m"
_YELLOW = "\x1b[33m"
_GREEN = "\x1b[32m"
_RED = "\x1b[31m"
def _color(text: str, code: str) -> str:
if _NO_COLOR:
return text
return f"{code}{text}{_RESET}"
# Configure logging to output DEBUG-WARNING to the stdout
# and ERROR-CRITICAL to the stderr.
# We need a custom filter for the stdout stream
# so it won't print anything higher than WARNING.
class _MaxLevelFilter(logging.Filter):
def __init__(self, max_level):
super().__init__()
self.max_level = max_level
def filter(self, record):
return record.levelno <= self.max_level
def _build_log_config(loggers, _filter_ref=None) -> dict:
if _filter_ref is None:
_filter_ref = "vc_init_dev._MaxLevelFilter"
return {
"version": 1,
"disable_existing_loggers": False,
"filters": {
"max_warning": {
"()": _filter_ref,
"max_level": logging.WARNING,
}
},
"handlers": {
"stdout": {
"class": "logging.StreamHandler",
"stream": "ext://sys.stdout",
"filters": ["max_warning"],
},
"stderr": {
"class": "logging.StreamHandler",
"stream": "ext://sys.stderr",
"level": "ERROR",
},
},
"loggers": loggers,
"root": {
"handlers": ["stdout", "stderr"],
"level": "INFO",
},
}
def _setup_server_log_routing(logger_name=None):
loggers = {}
if logger_name:
loggers[logger_name] = {
"handlers": ["stdout", "stderr"],
"level": "INFO",
"propagate": False,
}
logging.config.dictConfig(
_build_log_config(
loggers=loggers,
_filter_ref=_MaxLevelFilter,
),
)
def _build_uvicorn_log_config(default_fmt=None, access_fmt=None) -> dict:
try:
from uvicorn.config import LOGGING_CONFIG # type: ignore
uvicorn_fmts = LOGGING_CONFIG["formatters"]
except ImportError:
uvicorn_fmts = {
"default": {
"()": "uvicorn.logging.DefaultFormatter",
"fmt": "%(levelprefix)s %(message)s",
"use_colors": None,
},
"access": {
"()": "uvicorn.logging.AccessFormatter",
"fmt": '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
},
}
cfg = _build_log_config(
loggers={
"uvicorn": {
"handlers": ["stdout", "stderr"],
"level": "INFO",
"propagate": False,
},
"uvicorn.error": {"level": "INFO"},
"uvicorn.access": {
"handlers": ["access"],
"level": "INFO",
"propagate": False,
},
},
)
if default_fmt is None:
default_fmt = {**uvicorn_fmts["default"], "use_colors": not _NO_COLOR}
if access_fmt is None:
access_fmt = {**uvicorn_fmts["access"], "use_colors": not _NO_COLOR}
cfg["formatters"] = {"default": default_fmt, "access": access_fmt}
cfg["handlers"]["stdout"]["formatter"] = "default"
cfg["handlers"]["stderr"]["formatter"] = "default"
cfg["handlers"]["access"] = {
"class": "logging.StreamHandler",
"stream": "ext://sys.stdout",
"formatter": "access",
}
return cfg
def _build_hypercorn_log_config():
return _build_log_config(
loggers={
"hypercorn.error": {
"handlers": ["stdout", "stderr"],
"level": "INFO",
"propagate": False,
},
"hypercorn.access": {
"handlers": ["stdout"],
"level": "INFO",
"propagate": False,
},
},
)
def _patch_fastapi_cli_log_config():
try:
import fastapi_cli.utils.cli as _fcli # type: ignore
import fastapi_cli.cli as _fcli_cli # type: ignore
_orig_get_config = _fcli.get_uvicorn_log_config # to ensure it's there
_fcli_cli.get_uvicorn_log_config # to ensure it's there
except (ImportError, AttributeError):
return
def _get_routed_config():
orig = _orig_get_config()
return _build_uvicorn_log_config(
default_fmt={
"()": "fastapi_cli.utils.cli.CustomFormatter",
"fmt": orig["formatters"]["default"].get(
"fmt", "%(levelprefix)s %(message)s"
),
"use_colors": orig["formatters"]["default"].get("use_colors"),
},
access_fmt={
"()": "fastapi_cli.utils.cli.CustomFormatter",
"fmt": orig["formatters"]["access"].get(
"fmt",
"%(levelprefix)s %(client_addr)s - '%(request_line)s' %(status_code)s",
),
},
)
_fcli.get_uvicorn_log_config = _get_routed_config
# we need to patch the local binding as well
_fcli_cli.get_uvicorn_log_config = _get_routed_config
def _normalize_service_route_prefix(raw_prefix):
if not raw_prefix:
return ""
prefix = raw_prefix.strip()
if not prefix:
return ""
if not prefix.startswith("/"):
prefix = f"/{prefix}"
return "" if prefix == "/" else prefix.rstrip("/")
def _is_service_route_prefix_strip_enabled():
raw = os.environ.get("VERCEL_SERVICE_ROUTE_PREFIX_STRIP")
if not raw:
return False
return raw.lower() in ("1", "true")
_SERVICE_ROUTE_PREFIX = (
_normalize_service_route_prefix(os.environ.get("VERCEL_SERVICE_ROUTE_PREFIX"))
if _is_service_route_prefix_strip_enabled()
else ""
)
def _strip_service_route_prefix(path_value):
if not path_value:
path_value = "/"
elif not path_value.startswith("/"):
path_value = f"/{path_value}"
prefix = _SERVICE_ROUTE_PREFIX
if not prefix:
return path_value, ""
if path_value == prefix:
return "/", prefix
if path_value.startswith(f"{prefix}/"):
stripped = path_value[len(prefix) :]
return stripped if stripped else "/", prefix
return path_value, ""
# Pre-configure the root logger before user module import so that any log
# calls emitted at import time are routed to stdout/stderr correctly.
_setup_server_log_routing()
# ASGI/WSGI app detection
_MODULE_NAME = "__VC_DEV_MODULE_NAME__"
_ENTRY_ABS = "__VC_DEV_ENTRY_ABS__"
# Import user module by file path, matching vc_init.py's approach.
# https://docs.python.org/3/library/importlib.html#importing-a-source-file-directly
_spec = _importlib_util.spec_from_file_location(_MODULE_NAME, _ENTRY_ABS)
if _spec is None or _spec.loader is None:
raise RuntimeError(
f"Could not load module spec for '{_MODULE_NAME}' at {_ENTRY_ABS}"
)
_mod = _importlib_util.module_from_spec(_spec)
sys.modules[_MODULE_NAME] = _mod
_spec.loader.exec_module(_mod)
_user_app_name = (
"app"
if hasattr(_mod, "app")
else "application"
if hasattr(_mod, "application")
else None
)
if _user_app_name is None:
raise RuntimeError(
f"Missing 'app' or 'application' in module '{_MODULE_NAME}'. "
f"Define `app = ...` or `application = ...` in your entrypoint."
)
_user_app = getattr(_mod, _user_app_name, None)
def _get_positional_param_count(obj):
try:
sig = inspect.signature(obj)
return sum(
1
for p in sig.parameters.values()
if p.default is inspect.Parameter.empty
and p.kind
in (
inspect.Parameter.POSITIONAL_ONLY,
inspect.Parameter.POSITIONAL_OR_KEYWORD,
)
)
except (ValueError, TypeError):
return None
def _detect_app_type(app_obj):
# try .asgi attribute if it's available and is callable
asgi_attr = getattr(app_obj, "asgi", None)
if asgi_attr is not None and callable(asgi_attr):
return "asgi", asgi_attr
# For async detection, check the object itself first (works for plain
# functions/methods).
# For class instances iscoroutinefunction(obj) is False,
# so fall back to __call__.
is_async = inspect.iscoroutinefunction(app_obj)
if not is_async:
call_method = getattr(app_obj, "__call__", None)
if call_method is not None:
is_async = inspect.iscoroutinefunction(call_method)
# inspect.signature() already delegates to __call__ for class instances,
# and works directly on plain functions, so always inspect app_obj.
param_count = _get_positional_param_count(app_obj)
# ASGI (scope, receive, send)
if is_async and param_count == 3:
return "asgi", app_obj
# WSGI (environ, start_response)
if param_count == 2:
return "wsgi", app_obj
print(
_color(
f"Could not determine the application interface for '{_MODULE_NAME}:{_user_app_name}'\n"
f"Expected either:\n"
f" - An ASGI app: async callable(scope, receive, send)\n"
f" - A WSGI app: callable(environ, start_response)",
_RED,
),
file=sys.stderr,
)
sys.exit(1)
app_type, resolved_app = _detect_app_type(_user_app)
if app_type == "asgi":
_asgi_app = resolved_app
_wsgi_app = None
else:
_wsgi_app = resolved_app
_asgi_app = None
PUBLIC_DIR = "public"
# Prepare ASGI app
# Optional StaticFiles import; tolerate missing deps
StaticFiles = None
try:
from fastapi.staticfiles import StaticFiles as _SF # type: ignore
StaticFiles = _SF
except Exception:
try:
from starlette.staticfiles import StaticFiles as _SF # type: ignore
StaticFiles = _SF
except Exception:
StaticFiles = None
# Prepare static files app (if starlette/fastapi installed)
_static_app = None
if StaticFiles is not None:
try:
try:
_static_app = StaticFiles(directory=PUBLIC_DIR, check_dir=False)
except TypeError:
_static_app = StaticFiles(directory=PUBLIC_DIR)
except Exception:
_static_app = None
def _apply_service_route_prefix_to_scope(scope):
path_value, matched_prefix = _strip_service_route_prefix(scope.get("path", "/"))
if not matched_prefix:
return scope
updated_scope = dict(scope)
updated_scope["path"] = path_value
raw_path = scope.get("raw_path")
if isinstance(raw_path, (bytes, bytearray)):
try:
decoded = bytes(raw_path).decode("utf-8", "surrogateescape")
stripped_raw, _ = _strip_service_route_prefix(decoded)
updated_scope["raw_path"] = stripped_raw.encode("utf-8", "surrogateescape")
except Exception:
pass
existing_root = scope.get("root_path", "") or ""
if existing_root and existing_root != "/":
existing_root = existing_root.rstrip("/")
else:
existing_root = ""
updated_scope["root_path"] = f"{existing_root}{matched_prefix}"
return updated_scope
async def asgi_app(scope, receive, send):
effective_scope = _apply_service_route_prefix_to_scope(scope)
if _static_app is not None and effective_scope.get("type") == "http":
req_path = effective_scope.get("path", "/") or "/"
safe = _p.normpath(req_path).lstrip("/")
full = _p.join(PUBLIC_DIR, safe)
try:
base = _p.realpath(PUBLIC_DIR)
target = _p.realpath(full)
if (target == base or target.startswith(base + _p.sep)) and _p.isfile(
target
):
await _static_app(effective_scope, receive, send)
return
except Exception:
pass
await _asgi_app(effective_scope, receive, send)
# Prepare WSGI
def _is_safe_file(base_dir: str, target: str) -> bool:
try:
base = _p.realpath(base_dir)
tgt = _p.realpath(target)
return (tgt == base or tgt.startswith(base + os.sep)) and _p.isfile(tgt)
except Exception:
return False
def _static_wsgi_app(environ, start_response):
# Only handle GET/HEAD requests for static assets
if environ.get("REQUEST_METHOD", "GET") not in ("GET", "HEAD"):
return _not_found(start_response)
req_path = environ.get("PATH_INFO", "/") or "/"
safe = _p.normpath(req_path).lstrip("/")
full = _p.join(PUBLIC_DIR, safe)
if not _is_safe_file(PUBLIC_DIR, full):
return _not_found(start_response)
ctype, _ = mimetypes.guess_type(full)
headers = [("Content-Type", ctype or "application/octet-stream")]
try:
# For HEAD requests, send headers only
if environ.get("REQUEST_METHOD") == "HEAD":
start_response("200 OK", headers)
return []
with open(full, "rb") as f:
data = f.read()
headers.append(("Content-Length", str(len(data))))
start_response("200 OK", headers)
return [data]
except Exception:
return _not_found(start_response)
def _not_found(start_response):
start_response("404 Not Found", [("Content-Type", "text/plain; charset=utf-8")])
return [b"Not Found"]
def wsgi_app(environ, start_response):
path_info, matched_prefix = _strip_service_route_prefix(
environ.get("PATH_INFO", "/") or "/"
)
environ["PATH_INFO"] = path_info
if matched_prefix:
script_name = environ.get("SCRIPT_NAME", "") or ""
if script_name and script_name != "/":
script_name = script_name.rstrip("/")
else:
script_name = ""
environ["SCRIPT_NAME"] = f"{script_name}{matched_prefix}"
# Try static first; if 404 then delegate to user app
captured_status = ""
captured_headers = tuple()
body_chunks = []
def capture_start_response(status, headers, exc_info=None):
nonlocal captured_status, captured_headers
captured_status = status
captured_headers = tuple(headers)
# Return a writer that buffers the body
def write(chunk: bytes):
body_chunks.append(chunk)
return write
result = _static_wsgi_app(environ, capture_start_response)
# If static handler produced 200, forward its response
if captured_status.startswith("200 "):
# Send headers and any chunks collected
writer = start_response(captured_status, list(captured_headers))
for chunk in body_chunks:
writer(chunk)
return result
# Otherwise, delegate to user's WSGI app
return _wsgi_app(environ, start_response)
# Run dev server
if __name__ == "__main__":
# Development runner
#
# For WSGI: prefer Werkzeug, but fall back to stdlib wsgiref.
# For ASGI: prefer FastAPI CLI (dev command), then uvicorn, then hypercorn.
#
# The port is provided by the caller via the PORT environment variable.
host = "127.0.0.1"
_raw_port = os.environ.get("PORT")
if not _raw_port:
print(
_color("PORT environment variable is required.", _RED),
file=sys.stderr,
)
sys.exit(1)
port = int(_raw_port)
if app_type == "wsgi":
try:
from werkzeug.serving import run_simple # type: ignore
_setup_server_log_routing("werkzeug")
run_simple(host, port, wsgi_app, use_reloader=True, threaded=True)
except Exception:
print(
_color(
"Werkzeug not installed; falling back to wsgiref (no reloader).",
_YELLOW,
),
file=sys.stderr,
)
from wsgiref.simple_server import make_server
httpd = make_server(host, port, wsgi_app)
print(_color(f"Serving on http://{host}:{port}", _GREEN))
httpd.serve_forever()
else:
try:
from fastapi_cli.cli import dev as fastapi_dev # type: ignore
except ImportError:
fastapi_dev = None
if fastapi_dev is not None:
_patch_fastapi_cli_log_config()
fastapi_dev(
entrypoint="vc_init_dev:asgi_app", host=host, port=port, reload=True
)
sys.exit(0)
try:
import uvicorn # type: ignore
uvicorn.run(
"vc_init_dev:asgi_app",
host=host,
port=port,
use_colors=True,
reload=True,
log_config=_build_uvicorn_log_config(),
)
except Exception:
try:
import asyncio
from hypercorn.config import Config # type: ignore
from hypercorn.asyncio import serve # type: ignore
config = Config()
config.bind = [f"{host}:{port}"]
config.use_reloader = True
config.logconfig_dict = _build_hypercorn_log_config()
async def _run():
await serve(asgi_app, config)
asyncio.run(_run())
except Exception:
print(
_color(
'No ASGI server found. Please install either "uvicorn" or "hypercorn" (e.g. "pip install uvicorn").',
_RED,
),
file=sys.stderr,
)
sys.exit(1)
+7
-7
package.json
{
"name": "@vercel/python",
"version": "6.20.2",
"version": "6.21.0",
"main": "./dist/index.js",

@@ -9,4 +9,4 @@ "license": "Apache-2.0",

"dist",
"vc_init.py",
"vc_init_dev.py"
"templates",
"vc_init.py"
],

@@ -19,3 +19,3 @@ "repository": {

"dependencies": {
"@vercel/python-analysis": "0.8.2"
"@vercel/python-analysis": "0.9.0"
},

@@ -40,5 +40,5 @@ "devDependencies": {

"is-port-reachable": "3.1.0",
"@vercel/build-utils": "13.6.3",
"@vercel/python-runtime": "0.5.6",
"@vercel/error-utils": "2.0.3"
"@vercel/build-utils": "13.7.0",
"@vercel/error-utils": "2.0.3",
"@vercel/python-runtime": "0.6.0"
},

@@ -45,0 +45,0 @@ "scripts": {

-596
vc_init_dev.py
# Auto-generated template used by vercel dev (Python, ASGI/WSGI)
# Serves static files from PUBLIC_DIR before delegating to the user app.
import sys
import os
import inspect
import logging
import logging.config
from os import path as _p
from importlib import util as _importlib_util
import mimetypes
# Simple ANSI coloring. Respect NO_COLOR environment variable.
_NO_COLOR = "NO_COLOR" in os.environ
_RESET = "\x1b[0m"
_YELLOW = "\x1b[33m"
_GREEN = "\x1b[32m"
_RED = "\x1b[31m"
def _color(text: str, code: str) -> str:
if _NO_COLOR:
return text
return f"{code}{text}{_RESET}"
# Configure logging to output DEBUG-WARNING to the stdout
# and ERROR-CRITICAL to the stderr.
# We need a custom filter for the stdout stream
# so it won't print anything higher than WARNING.
class _MaxLevelFilter(logging.Filter):
def __init__(self, max_level):
super().__init__()
self.max_level = max_level
def filter(self, record):
return record.levelno <= self.max_level
def _build_log_config(loggers, _filter_ref=None) -> dict:
if _filter_ref is None:
_filter_ref = "vc_init_dev._MaxLevelFilter"
return {
"version": 1,
"disable_existing_loggers": False,
"filters": {
"max_warning": {
"()": _filter_ref,
"max_level": logging.WARNING,
}
},
"handlers": {
"stdout": {
"class": "logging.StreamHandler",
"stream": "ext://sys.stdout",
"filters": ["max_warning"],
},
"stderr": {
"class": "logging.StreamHandler",
"stream": "ext://sys.stderr",
"level": "ERROR",
},
},
"loggers": loggers,
"root": {
"handlers": ["stdout", "stderr"],
"level": "INFO",
},
}
def _setup_server_log_routing(logger_name=None):
loggers = {}
if logger_name:
loggers[logger_name] = {
"handlers": ["stdout", "stderr"],
"level": "INFO",
"propagate": False,
}
logging.config.dictConfig(
_build_log_config(
loggers=loggers,
_filter_ref=_MaxLevelFilter,
),
)
def _build_uvicorn_log_config(default_fmt=None, access_fmt=None) -> dict:
try:
from uvicorn.config import LOGGING_CONFIG # type: ignore
uvicorn_fmts = LOGGING_CONFIG["formatters"]
except ImportError:
uvicorn_fmts = {
"default": {
"()": "uvicorn.logging.DefaultFormatter",
"fmt": "%(levelprefix)s %(message)s",
"use_colors": None,
},
"access": {
"()": "uvicorn.logging.AccessFormatter",
"fmt": '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
},
}
cfg = _build_log_config(
loggers={
"uvicorn": {
"handlers": ["stdout", "stderr"],
"level": "INFO",
"propagate": False,
},
"uvicorn.error": {"level": "INFO"},
"uvicorn.access": {
"handlers": ["access"],
"level": "INFO",
"propagate": False,
},
},
)
if default_fmt is None:
default_fmt = {**uvicorn_fmts["default"], "use_colors": not _NO_COLOR}
if access_fmt is None:
access_fmt = {**uvicorn_fmts["access"], "use_colors": not _NO_COLOR}
cfg["formatters"] = {"default": default_fmt, "access": access_fmt}
cfg["handlers"]["stdout"]["formatter"] = "default"
cfg["handlers"]["stderr"]["formatter"] = "default"
cfg["handlers"]["access"] = {
"class": "logging.StreamHandler",
"stream": "ext://sys.stdout",
"formatter": "access",
}
return cfg
def _build_hypercorn_log_config():
return _build_log_config(
loggers={
"hypercorn.error": {
"handlers": ["stdout", "stderr"],
"level": "INFO",
"propagate": False,
},
"hypercorn.access": {
"handlers": ["stdout"],
"level": "INFO",
"propagate": False,
},
},
)
def _patch_fastapi_cli_log_config():
try:
import fastapi_cli.utils.cli as _fcli # type: ignore
import fastapi_cli.cli as _fcli_cli # type: ignore
_orig_get_config = _fcli.get_uvicorn_log_config # to ensure it's there
_fcli_cli.get_uvicorn_log_config # to ensure it's there
except (ImportError, AttributeError):
return
def _get_routed_config():
orig = _orig_get_config()
return _build_uvicorn_log_config(
default_fmt={
"()": "fastapi_cli.utils.cli.CustomFormatter",
"fmt": orig["formatters"]["default"].get(
"fmt", "%(levelprefix)s %(message)s"
),
"use_colors": orig["formatters"]["default"].get("use_colors"),
},
access_fmt={
"()": "fastapi_cli.utils.cli.CustomFormatter",
"fmt": orig["formatters"]["access"].get(
"fmt",
"%(levelprefix)s %(client_addr)s - '%(request_line)s' %(status_code)s",
),
},
)
_fcli.get_uvicorn_log_config = _get_routed_config
# we need to patch the local binding as well
_fcli_cli.get_uvicorn_log_config = _get_routed_config
def _normalize_service_route_prefix(raw_prefix):
if not raw_prefix:
return ""
prefix = raw_prefix.strip()
if not prefix:
return ""
if not prefix.startswith("/"):
prefix = f"/{prefix}"
return "" if prefix == "/" else prefix.rstrip("/")
def _is_service_route_prefix_strip_enabled():
raw = os.environ.get("VERCEL_SERVICE_ROUTE_PREFIX_STRIP")
if not raw:
return False
return raw.lower() in ("1", "true")
_SERVICE_ROUTE_PREFIX = (
_normalize_service_route_prefix(os.environ.get("VERCEL_SERVICE_ROUTE_PREFIX"))
if _is_service_route_prefix_strip_enabled()
else ""
)
def _strip_service_route_prefix(path_value):
if not path_value:
path_value = "/"
elif not path_value.startswith("/"):
path_value = f"/{path_value}"
prefix = _SERVICE_ROUTE_PREFIX
if not prefix:
return path_value, ""
if path_value == prefix:
return "/", prefix
if path_value.startswith(f"{prefix}/"):
stripped = path_value[len(prefix) :]
return stripped if stripped else "/", prefix
return path_value, ""
# Pre-configure the root logger before user module import so that any log
# calls emitted at import time are routed to stdout/stderr correctly.
_setup_server_log_routing()
# ASGI/WSGI app detection
_MODULE_NAME = "__VC_DEV_MODULE_NAME__"
_ENTRY_ABS = "__VC_DEV_ENTRY_ABS__"
# Import user module by file path, matching vc_init.py's approach.
# https://docs.python.org/3/library/importlib.html#importing-a-source-file-directly
_spec = _importlib_util.spec_from_file_location(_MODULE_NAME, _ENTRY_ABS)
if _spec is None or _spec.loader is None:
raise RuntimeError(
f"Could not load module spec for '{_MODULE_NAME}' at {_ENTRY_ABS}"
)
_mod = _importlib_util.module_from_spec(_spec)
sys.modules[_MODULE_NAME] = _mod
_spec.loader.exec_module(_mod)
_user_app_name = (
"app"
if hasattr(_mod, "app")
else "application"
if hasattr(_mod, "application")
else None
)
if _user_app_name is None:
raise RuntimeError(
f"Missing 'app' or 'application' in module '{_MODULE_NAME}'. "
f"Define `app = ...` or `application = ...` in your entrypoint."
)
_user_app = getattr(_mod, _user_app_name, None)
def _get_positional_param_count(obj):
try:
sig = inspect.signature(obj)
return sum(
1
for p in sig.parameters.values()
if p.default is inspect.Parameter.empty
and p.kind
in (
inspect.Parameter.POSITIONAL_ONLY,
inspect.Parameter.POSITIONAL_OR_KEYWORD,
)
)
except (ValueError, TypeError):
return None
def _detect_app_type(app_obj):
# try .asgi attribute if it's available and is callable
asgi_attr = getattr(app_obj, "asgi", None)
if asgi_attr is not None and callable(asgi_attr):
return "asgi", asgi_attr
# For async detection, check the object itself first (works for plain
# functions/methods).
# For class instances iscoroutinefunction(obj) is False,
# so fall back to __call__.
is_async = inspect.iscoroutinefunction(app_obj)
if not is_async:
call_method = getattr(app_obj, "__call__", None)
if call_method is not None:
is_async = inspect.iscoroutinefunction(call_method)
# inspect.signature() already delegates to __call__ for class instances,
# and works directly on plain functions, so always inspect app_obj.
param_count = _get_positional_param_count(app_obj)
# ASGI (scope, receive, send)
if is_async and param_count == 3:
return "asgi", app_obj
# WSGI (environ, start_response)
if param_count == 2:
return "wsgi", app_obj
print(
_color(
f"Could not determine the application interface for '{_MODULE_NAME}:{_user_app_name}'\n"
f"Expected either:\n"
f" - An ASGI app: async callable(scope, receive, send)\n"
f" - A WSGI app: callable(environ, start_response)",
_RED,
),
file=sys.stderr,
)
sys.exit(1)
app_type, resolved_app = _detect_app_type(_user_app)
if app_type == "asgi":
_asgi_app = resolved_app
_wsgi_app = None
else:
_wsgi_app = resolved_app
_asgi_app = None
PUBLIC_DIR = "public"
# Prepare ASGI app
# Optional StaticFiles import; tolerate missing deps
StaticFiles = None
try:
from fastapi.staticfiles import StaticFiles as _SF # type: ignore
StaticFiles = _SF
except Exception:
try:
from starlette.staticfiles import StaticFiles as _SF # type: ignore
StaticFiles = _SF
except Exception:
StaticFiles = None
# Prepare static files app (if starlette/fastapi installed)
_static_app = None
if StaticFiles is not None:
try:
try:
_static_app = StaticFiles(directory=PUBLIC_DIR, check_dir=False)
except TypeError:
_static_app = StaticFiles(directory=PUBLIC_DIR)
except Exception:
_static_app = None
def _apply_service_route_prefix_to_scope(scope):
path_value, matched_prefix = _strip_service_route_prefix(scope.get("path", "/"))
if not matched_prefix:
return scope
updated_scope = dict(scope)
updated_scope["path"] = path_value
raw_path = scope.get("raw_path")
if isinstance(raw_path, (bytes, bytearray)):
try:
decoded = bytes(raw_path).decode("utf-8", "surrogateescape")
stripped_raw, _ = _strip_service_route_prefix(decoded)
updated_scope["raw_path"] = stripped_raw.encode("utf-8", "surrogateescape")
except Exception:
pass
existing_root = scope.get("root_path", "") or ""
if existing_root and existing_root != "/":
existing_root = existing_root.rstrip("/")
else:
existing_root = ""
updated_scope["root_path"] = f"{existing_root}{matched_prefix}"
return updated_scope
async def asgi_app(scope, receive, send):
effective_scope = _apply_service_route_prefix_to_scope(scope)
if _static_app is not None and effective_scope.get("type") == "http":
req_path = effective_scope.get("path", "/") or "/"
safe = _p.normpath(req_path).lstrip("/")
full = _p.join(PUBLIC_DIR, safe)
try:
base = _p.realpath(PUBLIC_DIR)
target = _p.realpath(full)
if (target == base or target.startswith(base + _p.sep)) and _p.isfile(
target
):
await _static_app(effective_scope, receive, send)
return
except Exception:
pass
await _asgi_app(effective_scope, receive, send)
# Prepare WSGI
def _is_safe_file(base_dir: str, target: str) -> bool:
try:
base = _p.realpath(base_dir)
tgt = _p.realpath(target)
return (tgt == base or tgt.startswith(base + os.sep)) and _p.isfile(tgt)
except Exception:
return False
def _static_wsgi_app(environ, start_response):
# Only handle GET/HEAD requests for static assets
if environ.get("REQUEST_METHOD", "GET") not in ("GET", "HEAD"):
return _not_found(start_response)
req_path = environ.get("PATH_INFO", "/") or "/"
safe = _p.normpath(req_path).lstrip("/")
full = _p.join(PUBLIC_DIR, safe)
if not _is_safe_file(PUBLIC_DIR, full):
return _not_found(start_response)
ctype, _ = mimetypes.guess_type(full)
headers = [("Content-Type", ctype or "application/octet-stream")]
try:
# For HEAD requests, send headers only
if environ.get("REQUEST_METHOD") == "HEAD":
start_response("200 OK", headers)
return []
with open(full, "rb") as f:
data = f.read()
headers.append(("Content-Length", str(len(data))))
start_response("200 OK", headers)
return [data]
except Exception:
return _not_found(start_response)
def _not_found(start_response):
start_response("404 Not Found", [("Content-Type", "text/plain; charset=utf-8")])
return [b"Not Found"]
def wsgi_app(environ, start_response):
path_info, matched_prefix = _strip_service_route_prefix(
environ.get("PATH_INFO", "/") or "/"
)
environ["PATH_INFO"] = path_info
if matched_prefix:
script_name = environ.get("SCRIPT_NAME", "") or ""
if script_name and script_name != "/":
script_name = script_name.rstrip("/")
else:
script_name = ""
environ["SCRIPT_NAME"] = f"{script_name}{matched_prefix}"
# Try static first; if 404 then delegate to user app
captured_status = ""
captured_headers = tuple()
body_chunks = []
def capture_start_response(status, headers, exc_info=None):
nonlocal captured_status, captured_headers
captured_status = status
captured_headers = tuple(headers)
# Return a writer that buffers the body
def write(chunk: bytes):
body_chunks.append(chunk)
return write
result = _static_wsgi_app(environ, capture_start_response)
# If static handler produced 200, forward its response
if captured_status.startswith("200 "):
# Send headers and any chunks collected
writer = start_response(captured_status, list(captured_headers))
for chunk in body_chunks:
writer(chunk)
return result
# Otherwise, delegate to user's WSGI app
return _wsgi_app(environ, start_response)
# Run dev server
if __name__ == "__main__":
# Development runner
#
# For WSGI: prefer Werkzeug, but fall back to stdlib wsgiref.
# For ASGI: prefer FastAPI CLI (dev command), then uvicorn, then hypercorn.
#
# The port is provided by the caller via the PORT environment variable.
host = "127.0.0.1"
_raw_port = os.environ.get("PORT")
if not _raw_port:
print(
_color("PORT environment variable is required.", _RED),
file=sys.stderr,
)
sys.exit(1)
port = int(_raw_port)
if app_type == "wsgi":
try:
from werkzeug.serving import run_simple # type: ignore
_setup_server_log_routing("werkzeug")
run_simple(host, port, wsgi_app, use_reloader=True, threaded=True)
except Exception:
print(
_color(
"Werkzeug not installed; falling back to wsgiref (no reloader).",
_YELLOW,
),
file=sys.stderr,
)
from wsgiref.simple_server import make_server
httpd = make_server(host, port, wsgi_app)
print(_color(f"Serving on http://{host}:{port}", _GREEN))
httpd.serve_forever()
else:
try:
from fastapi_cli.cli import dev as fastapi_dev # type: ignore
except ImportError:
fastapi_dev = None
if fastapi_dev is not None:
_patch_fastapi_cli_log_config()
fastapi_dev(
entrypoint="vc_init_dev:asgi_app", host=host, port=port, reload=True
)
sys.exit(0)
try:
import uvicorn # type: ignore
uvicorn.run(
"vc_init_dev:asgi_app",
host=host,
port=port,
use_colors=True,
reload=True,
log_config=_build_uvicorn_log_config(),
)
except Exception:
try:
import asyncio
from hypercorn.config import Config # type: ignore
from hypercorn.asyncio import serve # type: ignore
config = Config()
config.bind = [f"{host}:{port}"]
config.use_reloader = True
config.logconfig_dict = _build_hypercorn_log_config()
async def _run():
await serve(asgi_app, config)
asyncio.run(_run())
except Exception:
print(
_color(
'No ASGI server found. Please install either "uvicorn" or "hypercorn" (e.g. "pip install uvicorn").',
_RED,
),
file=sys.stderr,
)
sys.exit(1)
./dist/index.js

Sorry, the diff of this file is too big to display

dist/index.js

Sorry, the diff of this file is too big to display