DEV Community: Justin Lin

mdsh: Run Shell Scripts in Markdown Templates

Justin Lin — Fri, 25 Jul 2025 11:47:06 +0000

I have been using hledger as my primary personal accounting software for years. I love that I can manage my ledger in plaintext and even use Git to version control and backup.

But when it comes to generating reports, it often takes me time to figure out all the commands I need. Also, having a way to archive previous data is important. I used to write a shell script with all the report commands, and add a lot of echo statements to generate a markdown report. However, this approach is hard to read and makes the template difficult to maintain.

That's why I made mdsh, a markdown template engine written in Go, which allows you to execute shell scripts within Markdown. It allows you to use Go's template syntax in markdown, and puts the execution results in the generated output.

You can install it with Go CLI:

go install github.com/lancatlin/mdsh@latest

The First Template

Suppose we want to generate a system information report. You can define a markdown template system-info.md as follows:

# 💥 System Information Report for {{ sh "hostname" }}

* **Hostname**: {{ sh "hostname" }}
* **Username**: {{ sh "whoami" }}
* **Uptime**: {{ sh "uptime -p" }}
* **System**: {{ sh "uname -a" }}
* **CPU**: {{ sh "uname -m" }} — {{ sh "nproc" }} cores
* **IP Address**: {{ sh "hostname -I || ip a | grep inet" }}
* **Default Gateway**: {{ sh "ip route | grep default || netstat -rn | grep default" }}

Run the template:

mdsh system-info.md

It will render into:

# 💥 System Information Report for `fedora`

* **Hostname**: `fedora`
* **Username**: `wancat`
* **Uptime**: `up 1 hour, 13 minutes`
* **System**: `Linux fedora 6.15.6-200.fc42.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jul 10 15:22:32 UTC 2025 x86_64 GNU/Linux`
* **CPU**: `x86_64` — `16` cores
* **IP Address**: `172.26.198.115 2405:dc00:ec83:ec80:af9c:87ed:9bae:bd0d`
* **Default Gateway**: `default via 172.26.198.50 dev wlp1s0 proto dhcp src 172.26.198.115 metric 600`

It makes generating reports as easy as a breeze.

Different Template Functions

Currently, it supports 3 types of template functions, which change the format they generate to:

sh: puts the output in inline code
shell: puts the output in a

code block

raw: puts output without any decorations

Also, any functions and syntax supported in text/template are supported.

Custom Parameters from Command Line Arguments

The best part of mdsh is that it supports custom parameters, which means you can define the parameters you're going to use in the template, and pass them through command line arguments.

You can access the parameters through both template data and environment variables. So you can access these variables from the script with ease.

For example, I need to generate a monthly report for my ledger. I need to specify begin and end times for the report.

Define the template as follows:

---
params:
  b:
    required: true
  e:
    required: true
  f:
    default: examples/ledger.j
---
// The template body

As you can see in the code, you can define custom parameters in the params: section in the frontmatter. I defined 3 parameters: b, e, and f, which stand for begin, end, and file. (This follows the convention in hledger)

Then I can use those parameters in the template body.

# Finance Report from {{.b}} to {{.e}}

## Net Income:

{{ shell "hledger -f $f income -b $b -e $e --monthly" }}

I use {{ .b }} to access the parameter directly and put it in the heading. Then I can use $b in the command. All the parameters will be passed as environment variables to the executing shell. So you can access them very easily.

mdsh hledger_monthly.md -b 2011-01 -e 2011-02

It will generate as follows:

# Finance Report from 2011-01 to 2011-02
## Net Income:

```
Income Statement 2011-01

                        ||        Jan
=========================++============
Revenues                ||
-------------------------++------------
Income:Salary           ||  $2,000.00
-------------------------++------------
                        ||  $2,000.00
=========================++============
Expenses                ||
-------------------------++------------
Expenses:Auto           ||  $5,500.00
Expenses:Books          ||     $20.00
Expenses:Food:Groceries ||    $109.00
-------------------------++------------
                        ||  $5,629.00
=========================++============
Net:                    || $-3,629.00`
```

It can even generate usage for each template based on the frontmatter.

params:
  b:
    required: true
    usage: |
      Required.
      The begin time of report.
      Examples:
        2025-07-01
        Jul
  f:
    usage: The ledger file to parse
    default: ~/.hledger.journal

Run the help command:

$ mdsh hledger_monthly.md -h
Usage of params:
  -b string
        Required.
        The begin time of report.
        Examples:
          2025
          Jul
         (default "2011-01")
  -e string
        Required.
        The end time of report.
        Same format as -b
         (default "2011-02")
  -f string
        The ledger file to parse (default "examples/ledger.j")

Output Filename Template

The default setting is writing the output to stdout.
If you want to save it in a file. You can do so by specifying output: in frontmatter.

output: monthly_report_{{.b}}.md

When running mdsh hledger_monthly.md -b 2025-07 -e 2025-08, it saves the output to monthly_report_2025-07.md. It helps you remain the naming consistency with ease.

Not only parameters, you can also put shell script into it. For example:

output: sys-report-{{ raw "date --iso-8601" }}.md

Then it will write the result to sys-report-2025-07-25.md.

Use Case Ideas

mdsh has great potential in areas that require lots of shell scripting and documentation, like sysadmin, devops, and security. It's also good for creating documentation and tutorials, reducing the hassle of pasting command outputs over and over.

You can make it fit into your workflow by defining your own templates, with usage notes inside. Thanks to the rich features of Go's template engine, it allows huge extensibility. You can apply condition checks ({{ if }}, {{ with }}) or even loops.

Some potential usages are:

Sysadmin/DevOps: system snapshots, cluster health
Documentation: release notes, test results
Education: lab reports, tutorials
Security/Compliance: audits, vulnerability scans
Personal Finance: hledger, budget reports (what I'm using it for)

And many more waiting for you to discover!

If you found this project useful, please give me a star on GitHub 🌟

Side note: I developed the first version of mdsh within 2 hours at midnight while trying out Zed, and was impressed by its performance. I didn't use much AI for this project—sometimes you just need time and space to enjoy programming.

Originally posted on wancat.cc

How to Handle Conversation in Chatbot in Python

Justin Lin — Thu, 29 Jul 2021 08:03:00 +0000

When you develop a chatbot, sometimes for user experience, you cannot ask your user send messages like commands. For example, we want to build a guess number bot. We want the bot works like this:

user: guess

bot: From what number?

user:: 25

bot: To what number?

user: 100

bot: Guess a number between 25 to 100

user: 64

bot: too small

user: 91

bot: too large

......

user: 83

bot: Correct! You spent 6 times to guess this number.

However, the common way we dealing with requests in the backend is one-request-one-response. That would be a disaster to separate a lot of handlers from the conversation. Why? Think about how to store the states? In global variables? Or database? Or Redis? Once you ask users one more question, you need to change the schema of your state, and the code becomes more complex.

In the following, I will show you how to deal with conversations and write the handler in a simple and straightforward way like this:

    def guess(self):
        '''Game function'''
        min_value = self.ask_number('From what number?')
        max_value = self.ask_number('To what number?')
        secret = randint(min_value, max_value)
        msg = f'Guess a number between {min_value} to {max_value}'
        counter = 0
        while True:
            counter += 1
            answer = self.ask_number(msg)
            if answer > secret:
                msg = 'Too large'
            elif answer < secret:
                msg = 'Too small'
            else:
                break
        self.reply(f'You spent {counter} times to guess the secret number.')

I will write a LINE bot for example, but it doesn't matter what platform you develop to. I will use Django and it's okay if you use other frameworks.

Setup Environment

It's for setting up the bot, you can skip this if you know it.

Clone my repo:

git clone https://github.com/lancatlin/python-chatbot-context.git
cd python-chatbot-context
pipenv install
pipenv shell

Go to LINE Developers to create a bot. Issue the token and your secret, put them in a .env file.

LINE_TOKEN=YOUR_TOKEN
LINE_SECRET=YOUR_SECRET

Then start Django.

python manage.py migrate    # for first execution
python manage.py runserver

Use Ngrok or something similar to tunnel the localhost:8000 to a public endpoint, and register the URL to LINE Messaging API.

Idea Describe

The main idea is to block the command thread until another message is received. When the program receives the 'guess' command, it will be executed in the command thread. Once the program needs input from the user, it put a message in the room's "requests queue". Then when the message comes in at another thread, it checks the room's requests queue and puts the message in the responses queue if not empty.

Implement

We implement it as MessageQueue class:

# guess/message_queue.py
import queue
from threading import RLock
from .line import get_room


class RequestTimout(Exception):
    pass


class MessageQueue:
    __lock = RLock()
    __requests = {}
    __responses = {}

    @classmethod
    def create_if_not_exists(cls, room):
        '''Create the requests and responses queues for the room if not exists'''
        with cls.__lock:
            if room not in cls.__requests:
                cls.__requests[room] = queue.Queue(maxsize=1)

            if room not in cls.__responses:
                cls.__responses[room] = queue.Queue(maxsize=1)

    @classmethod
    def handle(cls, event):
        '''Handle the message, check whether there is room request for'''
        room = get_room(event)
        cls.create_if_not_exists(room)

        try:
            if not cls.__requests[room].empty():
                cls.__responses[room].put(event, timeout=1)
                cls.__requests[room].get()
                return True
            return False
        except queue.Empty:
            '''No request, ignore the message'''
            return False

    @classmethod
    def request(cls, room, timeout=30):
        '''Request a message, block until message comes in or timeout'''
        try:
            cls.create_if_not_exists(room)

            cls.__requests[room].put_nowait(True)
            return cls.__responses[room].get(timeout=timeout)

        except queue.Empty:
            MessageQueue.clear(room)
            raise RequestTimout

    @classmethod
    def clear(cls, room):
        '''Clear the requests'''
        cls.create_if_not_exists(room)
        try:
            cls.__requests[room].get_nowait()
        except queue.Empty:
            pass

With this, we can implement our guess app very easily.

# guess/guess.py
from .message_queue import MessageQueue, RequestTimout
from .line import reply_text, get_room, get_msg
from random import randint


class Guess:
    '''Guess handle a guess number game'''

    def __init__(self, event):
        self.event = event
        try:
            self.guess()
        except RequestTimout:
            self.reply('Timeout')

    def guess(self):
        '''Game function'''
        min_value = self.ask_number('From what number?')
        max_value = self.ask_number('To what number?')
        secret = randint(min_value, max_value)
        msg = f'Guess a number between {min_value} to {max_value}'
        counter = 0
        while True:
            counter += 1
            answer = self.ask_number(msg)
            if answer > secret:
                msg = 'Too large'
            elif answer < secret:
                msg = 'Too small'
            else:
                break
        self.reply(f'You spent {counter} times to guess the secret number.')

    def ask(self, *msg):
        '''Ask a question to current user'''
        self.reply(*msg)
        self.event = MessageQueue.request(get_room(self.event))
        return get_msg(self.event)

    def ask_number(self, *msg):
        '''Ask a number, if not number, ask again'''
        try:
            content = self.ask(*msg)
            return int(content)
        except ValueError:
            return self.ask_number('Please input an integer.', *msg)

    def reply(self, *msg):
        '''Reply words to user'''
        reply_text(self.event, *msg)

You can see the main function is straightforward, with only 17 lines of code. What's more, it can handle multiple user inputs at the same time.

Get full code on GitHub.

Special thanks to YukinaMochizuki for giving me the initial idea from his Notion bot project.

Original posted on wancat.cc

How to Redirect Stdout to Streaming Response in Django

Justin Lin — Tue, 25 May 2021 07:41:08 +0000

Sometimes we need to execute some long tasks at the backend, and the tasks are complicated and error-prone. So we hope users can see the real-time console log. Thus we need to redirect the stdout in our functions to the user's browser.

Given a function like the following. How to see the stdout in real-time in the browser?

import time


def job(times):
    for i in range(times):
        print(f'Task #{i}')
        time.sleep(1)
    print('Done')
    time.sleep(0.5)

Streaming Response

Normally, responses are sent after all the data has been collected. However, sometimes we can't wait until the data is prepared. In this case, we'll use streaming response. In Django, it's StreamingHttpResponse. I'll call it SHR in the following article. StreamingHttpResponse accepts an iterator for input. It sends the value each time getting a new value from the iterator. To use it, we only need to implement an iterator function. It will send the value from yield to the user's browser in real-time.

# Example of StreamingHttpResponse

from django.http.response import StreamingHttpResponse
def example():
    for i in range(5):
        # Add <br> to break line in browser
        yield f'{i}<br>'

def stream(request):
    return StreamingHttpResponse(example())

Output (in browser):

Threading

As we execute the task and stream output at the same time, we need concurrence. In Python, there are multiple choices, like threading, multiprocessing, etc. I'll use threading in this article because it is easier for me.

# Example of threading

from threading import Thread
import time

def example(times):
    for i in range(times):
        print(i)
        time.sleep(1)

# Create Thread
thread = Thread(target=example, args=(5,))

# Start Thread
thread.start()

time.sleep(2)
print("This is printed in the main thread")

# Waiting thread to be done
thread.join()

Output:

0
1
This is printed in the main thread
2
3
4

Redirect Stdout

To change the location where Python print, we need to change sys. stdout. It accepts any File-like object. Specifically, we need to define an object with the write method.

# Example of redirect stdout
import sys

class Printer:
    def __init__(self):
        self.contents = []

    def write(self, value):
        self.contents.append(value)

printer = Printer()
sys.stdout = printer

print('This should be saved in printer')

sys.stdout = sys.__stdout__

print('This should be printed to stdout')

print(printer.contents)

Output:

This should be printed to stdout
['This should be saved in printer', '\n']

Implement Redirecting Stdout to Streaming Response

Environment

Python 3.8.5

Django 3.2

First, create Django Project

pip install django
django-admin startproject console_streaming
cd console_streaming
python manage.py startapp web

Install web

# console_streaming/settings.py

INSTALLED_APPS = [
    ...
    # Add web
    'web',
]

Create a view

# web/views.py

def stream(request):
    # Implement later
    pass

Bind to URL

# console_streaming/urls.py

from django.urls import path
from web import views

urlpatterns = [
    path('stream/', views.stream),
]

Testing Task

This is the testing function we're going to use. It will repeat printing a line and waiting a second for n times, and then print "Done".

# web/views.py
import time


def job(times):
    for i in range(times):
        print(f'Task #{i}')
        time.sleep(1)
    print('Done')
    time.sleep(0.5)

Printer class

We implement a Printer class to handle stdout, and we'll only use one instance in the whale program life cycle. Because sys.stdout does not thread-specific, if we use different stdout in different requests, one would grab stdout from another. So I use a dictionary to store the queue for different threads, and use current_thread() to identify and pick the right queue. If the current thread hasn't registered to Printer, use the default stdout.

# web/views.py
from queue import Queue
from threading import current_thread
import sys


class Printer:
    def __init__(self):
        self.queues = {}

    def write(self, value):
        '''handle stdout'''
        queue = self.queues.get(current_thread().name)
        if queue:
            queue.put(value)
        else:
            sys.__stdout__.write(value)

    def flush(self):
        '''Django would crash without this'''
        pass

    def register(self, thread):
        '''register a Thread'''
        queue = Queue()
        self.queues[thread.name] = queue
        return queue

    def clean(self, thread):
        '''delete a Thread'''
        del self.queues[thread.name]

# Initialize a Printer instance
printer = Printer()
sys.stdout = printer

### Streamer class

Next, we're going to implement concurrent execution and streaming response by a Streamer class. It will initialize a thread, and register it to the printer to get a queue. Then it repeats to read the value from queue and yield to response until the thread ends.

from threading import Thread


class Steamer:
    def __init__(self, target, args):
        self.thread = Thread(target=target, args=args)
        self.queue = printer.register(self.thread)

    def start(self):
        self.thread.start()
        print('This should be stdout')
        while self.thread.is_alive():
            try:
                item = self.queue.get_nowait()
                yield f'{item}<br>'
            except Empty:
                pass
        yield 'End'
        printer.clean(self.thread)

def stream(request):
    streamer = Steamer(job, (10,))
    return StreamingHttpResponse(streamer.start())

Run Django server

$ python manage.py runserver

Open http://localhost:8000/stream/

Then you can see

Each time you make a request, you can see one output in the terminal.

This should be stdout

Full views.py

from django.http.response import StreamingHttpResponse
from queue import Queue, Empty
from threading import Thread, current_thread
import time
import sys


class Printer:
    def __init__(self):
        self.queues = {}

    def write(self, value):
        '''handle stdout'''
        queue = self.queues.get(current_thread().name)
        if queue:
            queue.put(value)
        else:
            sys.__stdout__.write(value)

    def flush(self):
        '''Django would crash without this'''
        pass

    def register(self, thread):
        '''register a Thread'''
        queue = Queue()
        self.queues[thread.name] = queue
        return queue

    def clean(self, thread):
        '''delete a Thread'''
        del self.queues[thread.name]


printer = Printer()
sys.stdout = printer


class Steamer:
    def __init__(self, target, args):
        self.thread = Thread(target=target, args=args)
        self.queue = printer.register(self.thread)

    def start(self):
        self.thread.start()
        print('This should be stdout')
        while self.thread.is_alive():
            try:
                item = self.queue.get_nowait()
                yield f'{item}<br>'
            except Empty:
                pass
        yield 'End'
        printer.clean(self.thread)


def job(times):
    for i in range(times):
        print(f'Task #{i}')
        time.sleep(1)
    print('Done')
    time.sleep(0.5)


def stream(request):
    streamer = Steamer(job, (10,))
    return StreamingHttpResponse(streamer.start())

Complete code on GitHub

Original published at wancat.cc