Skip to content

Instantly share code, notes, and snippets.

View thorin-schiffer's full-sized avatar
✈️
ProductFlight

Thorin Schiffer thorin-schiffer

✈️
ProductFlight
View GitHub Profile
@thorin-schiffer
thorin-schiffer / cdk_moto.py
Created May 13, 2022 14:41
CDK/Cloudformation and moto
@pytest.fixture(scope="session")
def cdk(ssm_mock, s3_mock, sqs_mock):
with open(Path(os.path.dirname(__file__)) / "resources" / "template.yaml") as f:
template = f.read()
parsed = yaml.load(template)
assets_bucket_name = None
lambda_zips = {}
for name, spec in parsed["Resources"].items():
spec["Properties"]["ApiKeyRequired"] = False
spec["Properties"]["StageName"] = "test"
@thorin-schiffer
thorin-schiffer / test_cdk.py
Created May 13, 2022 14:14
Testing CDK config with pytest
def get_single_properties(element):
return list(element.values())[0]
def has_role_perm(role, perm):
value = get_single_properties(role)
for prop in value["Properties"]["ManagedPolicyArns"]:
if perm in str(prop):
return True
@thorin-schiffer
thorin-schiffer / cdk_lambda.py
Last active May 13, 2022 14:06
CDK Stack creation and lambda in it
import os
import subprocess
from pathlib import Path
from aws_cdk import (
Stack,
aws_lambda as lambda_,
Duration,
aws_sqs,
aws_lambda_event_sources,
{
"AWSTemplateFormatVersion" : "2010-09-09",
"Description" : "AWS CloudFormation Sample Template DynamoDB_Table: This template demonstrates the creation of a DynamoDB table. **WARNING** This template creates an Amazon DynamoDB table. You will be billed for the AWS resources used if you create a stack from this template.",
"Parameters" : {
"HashKeyElementName" : {
"Description" : "HashType PrimaryKey Name",
"Type" : "String",
"AllowedPattern" : "[a-zA-Z0-9]*",
@thorin-schiffer
thorin-schiffer / signals_to_triggers.py
Created December 31, 2021 15:29
Django-plpy signals to triggers
# Before:
from django.dispatch import receiver
from django.db.models.signals import post_save
from django.contrib.auth.models import User
@receiver(post_save, sender=User)
def send_mail(sender, instance, **kwargs):
instance.send_mail()
@thorin-schiffer
thorin-schiffer / bulk.py
Created December 31, 2021 15:27
Django-plpy bulk operations
from django_plpy.installer import pltrigger
from tests.books.models import Book
@pltrigger(event="UPDATE", when="BEFORE", model=Book)
def pl_update_amount(new: Book, old: Book, td, plpy):
# don't use save method here, it will kill the database because of recursion
new.amount_stock += 10
@thorin-schiffer
thorin-schiffer / triggers_orm.py
Created December 31, 2021 15:23
Django-plpy triggers with ORM
from django_plpy.installer import pltrigger
from django.db.models import Model, CharField, IntegerField
class Book(Model):
name = CharField(max_length=10)
amount_stock = IntegerField(default=20)
amount_sold = IntegerField(default=10)
@thorin-schiffer
thorin-schiffer / triggers.py
Created December 31, 2021 15:22
Django-plpy triggers without using ORM in them
from django_plpy.installer import pltrigger
@pltrigger(event="INSERT", when="BEFORE", table="books_book")
def pl_trigger(td, plpy):
# mind triggers don't return anything
td["new"]["name"] = td["new"]["name"] + "test"
td["new"]["amount_sold"] = plpy.execute("SELECT count(*) FROM books_book")[0][
"count"
@thorin-schiffer
thorin-schiffer / lookups.py
Created December 31, 2021 15:20
Django-plpy custom ORM lookups
from django_plpy.installer import plfunction
from django.db.models import Transform
from django.db.models import IntegerField
from tests.books.models import Book
@plfunction
def plsquare(a: int) -> int:
return a * a
class PySquare(Transform):
@thorin-schiffer
thorin-schiffer / annotations.py
Created December 31, 2021 15:20
Django-plpy annotations
from django.db.models import F, Func
from tests.books.models import Book
Book.objects.annotate(
max_value=Func(F("amount_sold"), F("amount_stock"), function="pl_max")
)