Decorate that un-registers a job if the server does not match any of the allowed servers. :param allowed_servers: List of allowed servers. Can include `*` for globbing, eg "ol-web*"
(allowed_servers: list[str], scheduler: AsyncIOScheduler)
| 71 | |
| 72 | |
| 73 | def limit_server(allowed_servers: list[str], scheduler: AsyncIOScheduler): |
| 74 | """ |
| 75 | Decorate that un-registers a job if the server does not match any of the allowed servers. |
| 76 | |
| 77 | :param allowed_servers: List of allowed servers. Can include `*` for globbing, eg "ol-web*" |
| 78 | """ |
| 79 | |
| 80 | def decorator(func): |
| 81 | hostname = os.environ.get("HOSTNAME") |
| 82 | assert hostname |
| 83 | server = hostname |
| 84 | if hostname.endswith(".us.archive.org"): |
| 85 | server = hostname.split(".")[0] |
| 86 | |
| 87 | if not any(fnmatch.fnmatch(server, pattern) for pattern in allowed_servers): |
| 88 | scheduler.remove_job(func.__name__) |
| 89 | return func |
| 90 | |
| 91 | return decorator |
nothing calls this directly
no outgoing calls
no test coverage detected