codemation

import asyncio, concurrent, os
            
def ping(host, count=2, timeout=2):
    response = os.system(f'ping -c {count} -W {timeout} {host} > /dev/null')
    status = 'success' if response == 0 else 'fail'
    return {host: status}

async def main(hosts, loop, **kw):
    args = [2, 2]
    if 'timeout' in kw:

codemation

# The following assumes run(sever) was called with an existing fastapi router as input var
def run(server):
    import asyncio, uuid
    from fastapi.websockets import WebSocket
    from fastapi.testclient import TestClient



    server.clients = {}

codemation

import asyncio, uuid
from fastapi.websockets import WebSocket
from fastapi.testclient import TestClient
from fastapi import FastAPI
import logging as log
log.basicConfig()

server = FastAPI()

server.clients = {}

codemation

from fastapi import FastAPI
from easyrpc.server import EasyRpcServer
from aiopyql.data import Database

server = FastAPI()

@server.on_event('startup')
async def db_setup():
    # Rpc Server
    db_server = await EasyRpcServer.create(

codemation

from easyrpc.proxy import EasyRpcProxy
from fastapi import FastAPI
from aiopyql.data import Database

server = FastAPI()

@server.on_event('startup')
async def setup():
    server.data = {}

codemation

# Cron Pattern
import asyncio
from fastapi import FastAPI

server = FastAPI()

server.cron_jobs = asyncio.Queue()

def cron(interval: int):
    def cron_setup(f):

codemation

import asyncio, uuid, time
import subprocess
from fastapi import FastAPI
from easyrpc.server import EasyRpcServer

server = FastAPI()

work_queue = asyncio.Queue()
work_results = {}

codemation

# subprocess job template for easyrpc_job_runner
import asyncio, sys, json
from easyrpc.proxy import EasyRpcProxy

async def job(reuest_id, payload):
    proxy = await EasyRpcProxy.create(
        '0.0.0.0', 
        8690, 
        '/ws/jobs', 
        server_secret='abcd1234',

codemation

# job manager to be used to add jobs / get results from an easyrpc job runner
import asyncio, sys, json
from easyrpc.proxy import EasyRpcProxy

async def get_proxy():
    proxy = await EasyRpcProxy.create(
        '0.0.0.0', 
        8690, 
        '/ws/jobs',
        server_secret='abcd1234',

codemation

# Backup
docker exec CONTAINER /usr/bin/mysqldump -u root --password=root DATABASE > backup.sql

# Restore
cat backup.sql | docker exec -i CONTAINER /usr/bin/mysql -u root --password=root DATABASE

# Restore into Kubernetes
cat backup.sql | kubectl exec -n <destination_namespace> -i <destination_pod> -- mysql -u root --password='password' <destination_database>