Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use stdout_formatted method in localstack

Best Python code snippet using localstack_python

admin.py

Source:admin.py

1import asyncio2import contextlib3import io4import json5import pprint6import textwrap7import time8import traceback9import discord10from discord.ext import commands11def format_codeblock(content):12    if not str(content):13        return None14    formatted = pprint.pformat(content, width=51, compact=True)15    if len(formatted) > 1014:16        formatted = formatted[:1013] + 'â¦'17    return f'```py\n{formatted}\n```'18def get_files(*contents):19    files = []20    for content in contents:21        if len(pprint.pformat(content, width=51, compact=True)) > 1014:22            files.append(discord.File(io.StringIO(pprint.pformat(content, width=128))))23    return files24class Admin(commands.Cog, command_attrs={"hidden": True}):25    def __init__(self, bot: commands.Bot):26        self.bot = bot27        self._last_eval_value = None28    def cog_check(self, ctx):29        return self.bot.is_owner(ctx.author)30    @commands.command(aliases=['.'])31    async def eval(self, ctx, *, code: str):32        if code.startswith('```'):33            code = '\n'.join(code.splitlines()[1:-1])34        else:35            code = f"return {code.strip('` ')}"36        code = 'async def func():\n' + textwrap.indent(code, '  ')37        code_return = '<empty>'38        code_stdout = io.StringIO()39        env = {'_': self._last_eval_value, 'b': self.bot}40        env.update(globals())41        env.update(locals())42        try:43            exec_time = time.perf_counter()44            exec(code, env)45            with contextlib.redirect_stdout(code_stdout):46                code_return = await env['func']()47        except:  # noqa: E72248            return_formatted = format_codeblock(code_return)49            stdout_formatted = format_codeblock(code_stdout.getvalue())50            traceback_formatted = format_codeblock(traceback.format_exc(-1))51            embed = discord.Embed(52                color=0xfa5050,53                title=":x: error!",54                description=f"{(time.perf_counter()-exec_time)*1000:g}ms :clock2:"55            )56            embed.add_field(name='Traceback', value=traceback_formatted, inline=False)57            embed.add_field(name='Return', value=return_formatted, inline=False)58            if stdout_formatted:59                embed.add_field(name='Stdout', value=stdout_formatted, inline=False)60            await ctx.send(61                embed=embed,62                files=get_files(code_return, code_stdout.getvalue(), traceback.format_exc(-1))63            )64        else:65            return_formatted = format_codeblock(code_return)66            stdout_formatted = format_codeblock(code_stdout.getvalue())67            embed = discord.Embed(68                color=0x50fa50,69                title=":white_check_mark: Code evaluated",70                description=f"{(time.perf_counter()-exec_time)*1000:g}ms :clock2:"71            )72            embed.add_field(73                name='Return', value=return_formatted, inline=False74            )75            if stdout_formatted:76                embed.add_field(77                    name='Stdout', value=stdout_formatted, inline=False78                )79            await ctx.send(80                embed=embed,81                files=get_files(code_return, code_stdout.getvalue())82            )83            if code_return is not None:84                self._last_eval_value = code_return85    @commands.command(aliases=['h'])86    async def shell(self, ctx, *, command):87        command = '\n'.join(command.splitlines()[1:-1]) \88                    if command.startswith('```') \89                    else command.strip('` ')90        exec_time = time.perf_counter()91        proc = await asyncio.subprocess.create_subprocess_shell(92            command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE93        )94        stdout, stderr = await proc.communicate()95        stdout_formatted = format_codeblock(stdout.decode())96        stderr_formatted = format_codeblock(stderr.decode())97        if proc.returncode == 0:98            embed = discord.Embed(99                color=0x50fa50,100                title=':white_check_mark: Subprocess finished',101                description=f"{(time.perf_counter()-exec_time)*1000:g}ms :clock2:"102            )103        else:104            embed = discord.Embed(105                color=0xfa7050,106                title=f':x: Subprocess exited with returncode {proc.returncode}',107                description=f"{(time.perf_counter()-exec_time)*1000:g}ms :clock2:"108            )109        if stdout_formatted is None and stderr_formatted is None:110            embed.description = 'no output Â· ' + embed.description111        if stdout_formatted is not None:112            embed.add_field(name='Stdout:', value=stdout_formatted, inline=False)113        if stderr_formatted is not None:114            embed.add_field(name='Stderr:', value=stderr_formatted, inline=False)115        await ctx.send(embed=embed, files=get_files(stdout.decode(), stderr.decode()))116    @commands.command()117    async def reloadcfg(self, ctx):118        self.bot.config = json.load(open('config_defaults.json')) | json.load(open('config.json'))119        await ctx.message.add_reaction('\N{white heavy check mark}')120    @commands.command(aliases=['l'])121    async def loadexts(self, ctx, *exts):122        if len(exts) == 0:123            await ctx.send(', '.join(f'`{i}`' for i in self.bot.extensions.keys()))124            return125        errors = []126        for ext in exts:127            ext = f'bot.exts.{ext}' if not ext.startswith('bot.exts.') else f'{ext}'128            try:129                if ext in self.bot.extensions:130                    self.bot.reload_extension(ext)131                else:132                    self.bot.load_extension(ext)133            except Exception as exc:134                exc = repr(exc)135                if len(exc) > 200:136                    exc = exc[:200] + 'â¦'137                errors.append(traceback.format_exc())138        if errors:139            await ctx.send(file=discord.File(io.StringIO('\n'.join(errors)), 'errors.py'))140        await ctx.message.add_reaction('\N{white heavy check mark}')141def setup(bot):...

Tomita_parse.py

Source:Tomita_parse.py

1import subprocess2import pymongo3import json4import re5from pymongo import MongoClient6import tonality7def start(text):8    9    client = MongoClient(10        "mongodb+srv://Admin:12345@db1-9o4za.mongodb.net/db1?retryWrites=true&w=majority")11    db = client.newsDB12    13    14    toParse = text.encode('utf-8', "strict")15    with subprocess.Popen(["tomita-parser", "config.proto"], stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE) as proc:16        stdout, stderr = proc.communicate(toParse)17        stdout_formatted = stdout.decode("utf-8", "strict")18        result = re.search(r'Name = .*',stdout_formatted)19        if(result != None):20            tonality.getTonality([result.group(0)[7:]])21            result = { "text": result.group(0)[7:]}22            print(result)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.