andgineer

import os
from invoke import Collection
import importlib

tasks_dir = os.path.dirname(os.path.abspath(__file__))
ns = Collection()

for filename in os.listdir(tasks_dir):
    if filename.endswith(".py"):
        module_name = os.path.splitext(filename)[0]

andgineer

    const firehoseRole = new iam.Role(this, 'FirehoseDeliveryRole', {
      assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
    });

    domain.grantWrite(firehoseRole);

    // Create a Kinesis Data Firehose delivery stream
    const deliveryStream = new firehose.CfnDeliveryStream(this, 'LogsDeliveryStream', {
      deliveryStreamType: 'DirectPut',
      openSearchDestinationConfiguration: {

andgineer

import socket
from testcontainers.core.container import DockerContainer


class DynamicPortDockerContainer(DockerContainer):
    def bind_available_port(self, container_port: int) -> int:
        """Find an available port on the host machine and bind it to the `container_port`.

        Return bound host port number.
        """

andgineer

def pytest_xdist_make_scheduler(config, log):
    """Create a custom scheduler for pytest-xdist."""
    return PriorityListScheduling(config, log)


"""Custom scheduler implementation that run tests from tests/resources/priority_tests.txt first.

For example that enables us to run long tests first and on separate nodes.
To get tests list sorted by duration we can use pytest --durations=200
"""

andgineer

aws servicediscovery list-services --filters Name=NAMESPACE_ID,Values=${NS_ID} --query "Services[?Name=='${SERVICE_NAME}'].Id"  --output text

andgineer

function s3du {
	readonly folder_to_scan=${1:?"The argument 's3://bucket/folder_to_scan/' must be specified."}
	
	for subfolder in $(aws s3 ls "${folder_to_scan}" | grep PRE | awk '{print $2}'); do 
		echo "${folder_to_scan}${subfolder}:" 
		aws s3 ls "${folder_to_scan}${subfolder}" --recursive \
			--human-readable \
			--summarize \ 
			| tail -n2 
	done

andgineer

require 'json'

dictionary_list = [
	{"Яндекс переводчик": "https://translate.yandex.ru/?lang=en-ru&text={query}"},
    {"Oxford Dictionary": "https://www.oxfordlearnersdictionaries.com/definition/english/{query}?q={query}"},
    {"Cambridge Dictionary": "https://dictionary.cambridge.org/dictionary/english-russian/{query}"}

]

script_filter_items = []

andgineer

def create_object(s3=None):
    session = get_session() if s3 is None else None
    with session.create_client(
        "s3"
    ) if s3 is None else contextlib.nullcontext() as s3_temp:
        return (s3 or s3_temp).get_object(Bucket=bucket, Key=key)

andgineer

    session = get_session()
    async with session.create_client("s3") as s3_temp:
        s3_object = await s3_temp.get_object(Bucket=bucket, Key=key)
        

andgineer

    session = get_session() if s3 is None else None
    async with session.create_client(
        "s3"
    ) if s3 is None else contextlib.AsyncExitStack() as s3_temp:
        s3_object = await (s3 or s3_temp).get_object(Bucket=bucket, Key=key)