config.py
配置文件的方式很多，starlette提供了一种.env文件的配置方式，实际像py，json，xml都是非常合适的配置文件。
class undefined:
pass
class EnvironError(Exception):
pass
Environ环境变量对象
class Environ(MutableMapping):
def __init__(self, environ: typing.MutableMapping = os.environ):
self._environ = environ
# 环境变量
self._has_been_read = set() # type: typing.Set[typing.Any]
# 已读项
def __getitem__(self, key: typing.Any) -> typing.Any:
self._has_been_read.add(key)
return self._environ.__getitem__(key)
def __setitem__(self, key: typing.Any, value: typing.Any) -> None:
if key in self._has_been_read:
raise EnvironError(
f"Attempting to set environ['{key}'], but the value has already been read."
)
self._environ.__setitem__(key, value)
def __delitem__(self, key: typing.Any) -> None:
if key in self._has_been_read:
raise EnvironError(
f"Attempting to delete environ['{key}'], but the value has already been read."
)
self._environ.__delitem__(key)
def __iter__(self) -> typing.Iterator:
return iter(self._environ)
def __len__(self) -> int:
return len(self._environ)
environ = Environ()
Config配置类
class Config:
def __init__(
self,
env_file: typing.Union[str, Path] = None,
environ: typing.Mapping[str, str] = environ,
) -> None:
"""
:param env_file: 配置文件
:param environ: 环境变量
"""
self.environ = environ
self.file_values = {} # type: typing.Dict[str, str]
if env_file is not None and os.path.isfile(env_file):
self.file_values = self._read_file(env_file)
def __call__(
self, key: str, cast: typing.Callable = None, default: typing.Any = undefined,
) -> typing.Any:
"""
:param key: 需要的键
:param cast: 回调封装类，可以自动将结果封装，如果为bool，则进行bool修正
:param default: 默认值
:return:
"""
return self.get(key, cast, default)
def get(
self, key: str, cast: typing.Callable = None, default: typing.Any = undefined,
) -> typing.Any:
# 获取配置
if key in self.environ:
value = self.environ[key]
return self._perform_cast(key, value, cast)
if key in self.file_values:
value = self.file_values[key]
return self._perform_cast(key, value, cast)
if default is not undefined:
return self._perform_cast(key, default, cast)
raise KeyError(f"Config '{key}' is missing, and has no default.")
def _read_file(self, file_name: typing.Union[str, Path]) -> typing.Dict[str, str]:
# 加载配置文件
file_values = {} # type: typing.Dict[str, str]
with open(file_name) as input_file:
for line in input_file.readlines():
line = line.strip()
if "=" in line and not line.startswith("#"):
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip("\"'")
file_values[key] = value
return file_values
def _perform_cast(
self, key: str, value: typing.Any, cast: typing.Callable = None,
) -> typing.Any:
# 执行修改
if cast is None or value is None:
return value
# 直接返回
elif cast is bool and isinstance(value, str):
mapping = {"true": True, "1": True, "false": False, "0": False}
value = value.lower()
if value not in mapping:
raise ValueError(
f"Config '{key}' has value '{value}'. Not a valid bool."
)
return mapping[value]
# 修正布尔值的书写方法
try:
return cast(value)
# 执行回调封装
except (TypeError, ValueError):
raise ValueError(
f"Config '{key}' has value '{value}'. Not a valid {cast.__name__}."
)
官方的使用示例
config = Config(".env")
DEBUG = config('DEBUG', cast=bool, default=False)
DATABASE_URL = config('DATABASE_URL', cast=databases.DatabaseURL)
SECRET_KEY = config('SECRET_KEY', cast=Secret)
ALLOWED_HOSTS = config('ALLOWED_HOSTS', cast=CommaSeparatedStrings)
templating.py
starlette使用的是jinja2模板，因为内容和框架关联不大，这里仅仅贴出来
class _TemplateResponse(Response):
media_type = "text/html"
def __init__(
self,
template: typing.Any,
context: dict,
status_code: int = 200,
headers: dict = None,
media_type: str = None,
background: BackgroundTask = None,
):
self.template = template
self.context = context
content = template.render(context)
# 渲染模板
super().__init__(content, status_code, headers, media_type, background)
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
request = self.context.get("request", {})
extensions = request.get("extensions", {})
if "http.response.template" in extensions:
await send(
{
"type": "http.response.template",
"template": self.template,
"context": self.context,
}
)
await super().__call__(scope, receive, send)
class Jinja2Templates:
"""
使用示例
templates = Jinja2Templates("templates")
return templates.TemplateResponse("index.html", {"request": request})
"""
def __init__(self, directory: str) -> None:
assert jinja2 is not None, "jinja2 must be installed to use Jinja2Templates"
self.env = self.get_env(directory)
def get_env(self, directory: str) -> "jinja2.Environment":
@jinja2.contextfunction
def url_for(context: dict, name: str, **path_params: typing.Any) -> str:
request = context["request"]
return request.url_for(name, **path_params)
loader = jinja2.FileSystemLoader(directory)
env = jinja2.Environment(loader=loader, autoescape=True)
env.globals["url_for"] = url_for
return env
def get_template(self, name: str) -> "jinja2.Template":
return self.env.get_template(name)
def TemplateResponse(
self,
name: str,
context: dict,
status_code: int = 200,
headers: dict = None,
media_type: str = None,
background: BackgroundTask = None,
) -> _TemplateResponse:
if "request" not in context:
raise ValueError('context must include a "request" key')
template = self.get_template(name)
return _TemplateResponse(
template,
context,
status_code=status_code,
headers=headers,
media_type=media_type,
background=background,
)
