fermayo

import importlib
from functools import wraps
from myproject.celery import app


@app.task
def call_async_task(obj_module_name, obj_class_name, obj_pk, obj_method, obj_args=None, obj_kwargs=None):
    model_class = getattr(importlib.import_module(obj_module_name), obj_class_name)
    obj = model_class.objects.get(pk=obj_pk)
    method = getattr(obj, obj_method)

fermayo

FROM alpine:latest AS cloudfront
RUN apk --no-cache add curl jq
RUN curl https://ip-ranges.amazonaws.com/ip-ranges.json | \
    jq -r '.prefixes[] | select(.service=="CLOUDFRONT") | .ip_prefix' | \
    xargs -I '{}' echo 'set_real_ip_from {};' > /cloudfront.conf && \
    echo 'real_ip_header X-Forwarded-For;' >> /cloudfront.conf && \
    echo 'real_ip_recursive on;' >> /cloudfront.conf

FROM nginx:latest
COPY --from=cloudfront /cloudfront.conf /etc/nginx/conf.d/cloudfront.conf

fermayo

import requests

def depaginate(initial_url, **kwargs):
    r = requests.get(initial_url, **kwargs)
    r.raise_for_status()
    yield r.json()
    while r.links.get('next'):
        r = requests.get(r.links.get('next'), **kwargs)
        r.raise_for_status()
        yield r.json()

fermayo

for NAMESPACE in $(kubectl get namespaces -o jsonpath='{.items[*].metadata.name}'); do
  kubectl -n $NAMESPACE delete job $(kubectl -n $NAMESPACE get jobs -o jsonpath='{.items[?(@.status.completionTime)].metadata.name}')
done