jasima/majuna
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

lists: mirror mappings version 1.1

Browse source 
adds s3_buckets
This commit is contained in:
Iain Learmonth 2022-05-11 10:38:27 +01:00
parent bac83dc6b4
commit f3a6a5413d
1 changed files with 22 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

32
app/lists/mirror_mapping.py
View file

@ -1,10 +1,12 @@
import builtins import builtins
from typing import Dict from typing import Dict, List
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from tldextract import extract from tldextract import extract
from app import Proxy from app import app
from app.models.base import Group
from app.models.mirrors import Proxy
class MMMirror(BaseModel): class MMMirror(BaseModel):
@ -20,19 +22,29 @@ class MirrorMapping(BaseModel):
mappings: Dict[str, MMMirror] = Field( mappings: Dict[str, MMMirror] = Field(
description="The domain name for the mirror" description="The domain name for the mirror"
) )
s3_buckets: List[str] = Field(
description="The names of all S3 buckets used for CloudFront logs"
)
class Config: class Config:
title = "Mirror Mapping Version 1" title = "Mirror Mapping Version 1.1"
def mirror_mapping(): def mirror_mapping():
return MirrorMapping(version="1.0", mappings={ return MirrorMapping(
d.url.lstrip("https://"): MMMirror( version="1.1",
origin_domain=d.origin.domain_name, mappings={
origin_domain_normalized=d.origin.domain_name.lstrip("www."), d.url.lstrip("https://"): MMMirror(
origin_domain_root=extract(d.origin.domain_name).registered_domain origin_domain=d.origin.domain_name,
) for d in Proxy.query.all() if d.url is not None origin_domain_normalized=d.origin.domain_name.lstrip("www."),
}).dict() origin_domain_root=extract(d.origin.domain_name).registered_domain
) for d in Proxy.query.all() if d.url is not None
},
s3_buckets=[
f"{app.config['GLOBAL_NAMESPACE']}-{g.group_name}-logs-cloudfront"
for g in Group.query.filter(Group.destroyed == None).all()
]
).dict()
if getattr(builtins, "__sphinx_build__", False): if getattr(builtins, "__sphinx_build__", False):