Forem: Azizul Haque Ananto

Elixir Pipe in Python v2

Azizul Haque Ananto — Tue, 27 Feb 2024 16:52:59 +0000

I had previously shared a cool way to chain functions. This is exactly same but with hacking the | (Or) operator ;)

import json
from dataclasses import dataclass
from functools import reduce

import requests


class Pipe:
    def __init__(self, val):
        self.val = val
        self.funcs = []

    def __or__(self, other):
        self.funcs.append(other)
        return self

    def __call__(self):
        return reduce(
            lambda x, func: func(x) if not isinstance(x, list) else func(*x),
            self.funcs,
            self.val,
        )


@dataclass
class Request:
    url: str
    method: str


def parse_request(req: str):
    return (Pipe(req) | json.loads | (lambda x: Request(**x)))()


def validate_url(req: Request):
    return req if req.url.startswith("http") else None


def validate_method(req: Request):
    return req if req.method in ["GET", "POST", "PUT", "DELETE"] else None


def get_content(req: Request):
    return requests.get(req.url).content


def check_content_type(content: str, content_type: str):
    return content if content.startswith(content_type) else None


pp = (
    Pipe("""{"url": "https://google.com", "method": "GET"}""")
    | parse_request
    | validate_url
    | validate_method
    | get_content
    | (lambda x: x.decode("utf-8"))
    | (lambda x: check_content_type(x, "<!doctype html>"))
)()

print(pp)

Let me know your thoughts!

**The cover image is taken from here

Elixir Pipe in Python

Azizul Haque Ananto — Mon, 09 May 2022 16:38:37 +0000

I really like the Pipe operator in Elixir. This is something from the future!

As I work with Python more often, I was trying to implement that pipe functionality in Python with few lines of codes.

from functools import reduce


def pipe(*fn_list):
    return reduce(
        lambda x, f: f(x) if not isinstance(x, list) else f(*x),
        fn_list,
    )

How to use it?

Example -

import json
from dataclasses import dataclass
import requests


@dataclass
class Request:
    url: str
    method: str


def parse_request(req: str):
    return pipe(
        req,
        json.loads,
        lambda x: Request(**x),
    )


def validate_url(req: Request):
    return req if req.url.startswith("http") else None


def validate_method(req: Request):
    return req if req.method in ["GET", "POST", "PUT", "DELETE"] else None


def get_content(req: Request):
    return requests.get(req.url).content


def check_content_type(content: str, content_type: str):
    return content if content.startswith(content_type) else None


print(
    pipe(
        """{"url": "https://google.com", "method": "GET"}""",
        parse_request,
        validate_url,
        validate_method,
        get_content,
        lambda content: content.decode("utf-8"),
        lambda content: check_content_type(content, "<!doctype html>"),
    )
)

You can use lambda and functions in pipe.

Notice the last function in pipe, check_content_type takes two arguments, the first one content is passed from the previous pipe function, so we take that in lambda argument and then call the check_content_type with that content and another argument.

Let me know your thoughts!

**The cover image is taken from here.

Zero: A fast and high performance Python RPC framework for building microservices or distributed servers

Azizul Haque Ananto — Sun, 03 Oct 2021 12:20:21 +0000

I have been working with Zero for couple of months and I can feel that Zero has potential to make the difference in writing microservices. Rather than thinking about boilerplate coding and REST services, you can focus on making things, writing business logic - this is the goal of Zero.

Let's take a look at a simple example -

Install it first

pip install zeroapi

server.py

from zero import ZeroServer

def echo(msg: str) -> str:
    return msg

async def hello_world() -> str:
    return "hello world"

if __name__ == "__main__":
    app = ZeroServer(port=5559)
    app.register_rpc(echo)
    app.register_rpc(hello_world)
    app.run()

This is it! The ZeroServer utilizes the cpu cores unlike other Python programs.

If you save run python3 server.py this will run several processes and distribute the tasks among them.

So you register a function here and return something. So this works like a RPC. The main difference with RPC is - the functions can take only one parameter, that we call a message in-short msg. There are several types we support now - most of basic types like int, float, str, bool, list, dict, tuple, set. And some typing types typing.List, typing.Tuple, typing.Dict, typing.Union, typing.Optional. We have a plan to support Pydantic 🙌

Now let's make a client for the server -

from zero import ZeroClient

zero_client = ZeroClient("localhost", 5559)

def echo():
    resp = zero_client.call("echo", "Hi there!")
    print(resp)

def hello():
    resp = zero_client.call("hello_world", None)
    print(resp)

if __name__ == "__main__":
    echo()
    hello()

Simple and usual. You need performance? We also support async 😃

import asyncio

from zero import AsyncZeroClient

zero_client = AsyncZeroClient("localhost", 5559)

async def echo():
    resp = await zero_client.call("echo", "Hi there!")
    print(resp)

async def hello():
    resp = await zero_client.call("hello_world", None)
    print(resp)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(echo())
    loop.run_until_complete(hello())

Run and play with these.

If you need to send two, three ... n parameters. You can just wrap in a list, send and unwrap the list in server. Like this -

server.py

def pythagoras(msg: typing.Tuple[int, int]) -> int:
    a, b = msg
    return a**2 + b**2

client.py

def get_pythagoras(a: int, b:int) -> int:
    return zero_server.call("pythagoras", (4, 5))

So you can send arbitrary number of arguments like this.

Are you lazy like me? Zero also generates client code! 🙌

If you run -

python -m zero.generate_client --host localhost --port 5559 --overwrite-dir ./my_client

This will generate a client code for you like this -

import typing  # remove this if not needed
from typing import List, Dict, Union, Optional, Tuple  # remove this if not needed
from zero import ZeroClient


zero_client = ZeroClient("localhost", 5559)


class RpcClient:
    def __init__(self, zero_client: ZeroClient):
        self._zero_client = zero_client

    def echo(self, msg: str) -> str:
        return self._zero_client.call("echo", msg)

    def hello_world(self, msg: str) -> str:
        return self._zero_client.call("hello_world", msg)

And you can use like this -

from my_client import RpcClient, zero_client

client = RpcClient(zero_client)

if __name__ == "__main__":
    client.echo("Hi there!")
    client.hello_world(None)

So try Zero and let me know 📩
Checkout the examples here - https://github.com/Ananto30/zero/tree/main/examples

If you like Zero please leave a star at GitHub: https://github.com/Ananto30/zero

And let's talk more in the comments 👇

Create your own blazing fast mock server with just a JSON file!

Azizul Haque Ananto — Wed, 19 May 2021 12:43:36 +0000

TLDR; https://github.com/Ananto30/mocker

The idea is really simple. Just create high-performing 🚀 mock servers using a json file that maps path to status code and response body.

This can be used to benchmark your microservices, that are dependant on other HTTP services.

This is a sample mock file -

{
  "/hello/worlds": {
    "statusCode": 200,
    "responseBody": {
      "message": "Hello worlds!",
      "data" : {
        "time": "now"
      },
      "worlds": [
        "cross origin world",
        "mars world",
        "moon world"
      ]
    }
  }
}

Then just run -

./mocker -addr <YOUR_HOST_AND_PORT> -file <MOCK_JSON_FILE_LOCATION>

In windows -

mocker.exe -addr <YOUR_HOST_AND_PORT> -file <MOCK_JSON_FILE_LOCATION>

Please report bugs and suggest new features and ideas. 👇

I have built an async to sync service using Vert.x event bus; please review the code and suggest better approaches

Azizul Haque Ananto — Sat, 03 Apr 2021 03:44:52 +0000

TLDR; Please visit here - https://github.com/Ananto30/vertx-async-gateway and review the code.

I have used ConcurrentHashMap to keep track of the requests. And assuming each async request should have a unique identifier that can be used to track it when we get the callback in a webhook.

Using this code, one can write new endpoints that will be calling third parties who are async (send the response in a callback URL). So when a call is received from a user, the response will be put in the ConcurrentHashMap against the identifier and block the request using the blocking handler of vertx (the response is not end()). Then a callback will be received in the callback endpoint and an event will be published in the bus. Another verticle will take the event and find it from the hashmap and end() the result to that specific request.

Is this code seems good enough for this purpose?

I have built a story pointing app with Svelte and socket.io

Azizul Haque Ananto — Thu, 18 Feb 2021 06:39:35 +0000

I am amazed by Svelte. This is super simple and easy. You just write and it works. Easiest store and state management and write less code. I just loved it.

So I built this voting app. This just ensures an unbiased environment to point a story. In agile we do this story pointing and get biased with others' opinion. Here we get a timer to vote and then see the result.

http://storypoint.dedsec.life/

GitHub - https://github.com/Ananto30/story-point

Please report any issue or bug and all reviews and feature requests are appreciated 😀

MongoDB Atlas Search: Super Simple, Very Powerful

Azizul Haque Ananto — Wed, 04 Nov 2020 08:01:49 +0000

I have built a Hadith search engine 🔍 called Ask Hadith using Atlas Search of MongoDB.

You can jump right into the code.
Or visit the article I have written here - https://azizulhaq-ananto.medium.com/mongodb-atlas-search-super-simple-very-powerful-6a2cdafa79d1

I have shared my journey with Atlas and the whole process from moving data to writing search queries. It was really to start with and loved the compound operator which can make rocket science in the pocket 😁

Please read the article 📖 and share your thoughts and if you have already tried Atlas Search, please let me know in the comments 👇
It would be great if you support Ask Hadith ❤️

Show off your games using this tool

Azizul Haque Ananto — Mon, 26 Oct 2020 09:26:27 +0000

I have built this dynamic 🖼 SVG generator that collects the 🕹 games you play from Steam and generates an image that can be embedded in any site (cause all sites use html 😅). You can also put it in ⓘ .md files (like readme)! For example in GitHub profile readme, look at mine -

Ananto30 / Ananto30

My profile with Steam stats card

👋 Hi, I'm Ananto

LinkedIn | Codersrank | Dev.to | Medium | Portfolio

🚀 Projects

Zero: A high performance and fast Python framework (RPC + PubSub) to build microservices. (WIP)
Ask Hadith: Hadith search engine. askhadith.com
Golpo: Social network. golpo.vercel.app (Need contributors 🙌)
KV Store: Redis based Config or Settings key-value store with UI and code generation.
Bazar: Online shop built with svelte. gg13.netlify.app (WIP)
Gmocker: Create a blazing fast mock server with just a JSON file! (Go)
Storypoint: Realtime story pointing (or T-shirt sizing) app. storypoint.onrender.com
News13: Minimalistic uncluttered latest news (of Bangladesh). news13.netlify.app
Cap-em: Resource limit tracker.
Loggable: A logging annotation library for reactive java.

🔬 POC's

Async to sync using
- Go channels - repo
- Vert.x - repo
A Fiber(Go) backend for a diary app - repo

🎮 My Steam games! (repo) -

🧑‍💻 My codersrank

…

View on GitHub

All you need is to make your Steam profile public. The steps are shown in the repo -

Ananto30 / steam-stat

🎮 Dynamically generate your Steam (game) stats card on SVG

Steam-stats

Steam stats svg generator. Create your own Steam stat dynamically and share anywhere.

This can be used with .md files!

You can put in your GitHub readme too 😃

Example 📝

For .md files -

![Travis](https://steam-stat.vercel.app/api?profileName=ananto)

For using as img property in html or md files -

<img src="https://steam-stat.vercel.app/api?profileName=ananto"/>

You need to provide the profile name in query parameter ?profileName=ananto.

Please ensure that your Steam profile is public!

How to get profile name or url

How to make profile public

Contribution

Contributions are welcomed ❤️

There are many things to add, so please feel free to discuss and improve the code quality

Inspired by github-readme-stats and natemoo-re

View on GitHub

I am attaching the following line in this blog post and it will dynamically generate the SVG every time someone visits this page!

![Travis](https://steam-stat.vercel.app/api?profileName=ananto)

Result -

(Obviously this has some caching to reduce spamming the Steam API 😛)

Please give it a ⭐️ star if you like it. 😃
Suggest any further improvements and contributions are also welcomed. 👏
Share your profile in the comment! 👇

An example Jooby app, Jooby is one of the fastest and performant framework in Java

Azizul Haque Ananto — Sat, 20 Jun 2020 07:56:58 +0000

TL;DR
Here is the repo - https://github.com/Ananto30/jooby-example-shortify

Jooby seems a very promising framework. It's easy and fast. I noticed the Techempower benchmark last week and found this amazing framework that supports Java and Kotlin.

History(!!) 🧐

I have been working with Webflux for around a year. Webflux is the implementation of reactive Java on top of Spring-boot framework. It is basically a non-blocking concept and reuses threads for concurrent connections, really good for modern services with lots of concurrent requests. But writing reactive code is not that fun, you have to maintain a chain and catching errors and no filtering and less readable code and so on... 😰 That's another talk.

So basically I was looking for another framework with the same performance but not reactive 😐 And I think Jooby is that one! 😃 And also it has the ability to run Event-loop, the non-blocking concept like Webflux where we can use the reactive java. Even the default mode supports both blocking and non-blocking in the same app! This is really cool!

Getting started

To begin with, it's really easy to setup a Jooby project, though it hasn't got starter like Spring-boot, but the jooby-cli is enough to bootstrap a new projects with varities of configurations. Easy way is to -

Download the jooby-cli (please refer to the doc's download, the link may change)
Add bin/jooby or bin/jooby.bat to the system path variable.
Set workspace jooby set -w ~/Source
Create a new project jooby> create myapp --gradle --mvc --docker This is how I created mine, you can find other options by jooby> help create.

You can find the details here - https://jooby.io/#getting-started.

Overview

I have created this app - https://github.com/Ananto30/jooby-example-shortify for exampling and maybe I will add more modules and functionalities to it for future tutorials. This is a url shortener app, uses redis to keep track of them. A very simple app, implemented in a spring like architecture.

I won't go deeper into every module or explaining all of my code. I will do that if you need me to, please comment for that. Mostly I will talk about the overall app structure.

I have used the mvc api of jooby. It's more like the spring-boot concept. My app has a controller, service, and some dto's. You can relate the architecture to spring-boot. All you need is to install the modules, there are a handful of modules already integrated to jooby and for other ones you just need to write an extension and register that in jooby's service registy, more like a bean in spring-boot, but with less magic, full control.

You can pass Context to any controller, handler, or service, just take it as an argument. There are filters, I have used them for request response logging, somewhat like the AOP. The error handler is really easy to integrate. And much more... I really loved the framework, it's easy with less magic!

Please check out my repository for a thorough idea of the app, and if you have any questions or idea or opinion or anything 😅 comment down below 👇

Deploy Django or Flask in 3 easy steps (in production)

Azizul Haque Ananto — Tue, 09 Jun 2020 19:04:53 +0000

No intro, let's jump to the steps.

1. Write a Dockerfile 😐 (why not)

This is my directory structure -

├── app/
│     .....
├── requirements.txt
├── Dockerfile
└── gunicorn_starter.sh

This is the Dockerfile -

FROM python:3.7.3-slim

WORKDIR /user/src

COPY ./requirements.txt ./
RUN pip3 install -r ./requirements.txt

COPY ./app ./app
COPY ./gunicorn_starter.sh ./

ENTRYPOINT ["./gunicorn_starter.sh"]

Things to remember -

Use small docker images like -slim or -alpine. -alpine is the lightest but in my case -alpine haven't got the gcc so I used -slim. (I could've install gcc manually in alpine)
Import requirements and install them before importing codes. Why? If we change our code, we won't need to install requirements in every build. Simple and easy way to reduce build times 🤗

2. Gunicorn server

Why Gunicorn? 🧐 -> Django or Flask is a framework, not server. So we need a server to serve our application built with the framework. Gunicorn is a WSGI supported server that can communicate with other application that supports WSGI like Flask or Django. WSGI is a gateway interface that matches the URI defined in the python application.

Now install gunicorn and add to the requirements.txt

pip install gunicorn
pip freeze > requirements.txt

This is the gunicorn_starter.sh file we have seen earlier -

gunicorn --chdir app wsgi:app -w 2 --threads 2 -b 0.0.0.0:8003

We show gunicorn our project directory with --chdir app. wsgi:app for Django apps and for Flask, we need the file name where we did app = Flask(), like this - <file_contains_app>:app. Change worker and thread as per your need. Too much workers actually makes a system slow, why? Because workers are processes, and they shares same CPU core. And we should know our threads right? 😅

3. Build & Deploy (Seek & Destroy ☠️)

Build the docker image -

docker build -t <your_tag> .

This image is production ready! Beleive it or not 😐

Run it 🤘 -

docker run --name <image_name> -p 8003:8003 -e DB_URI=<your_db_uri> <your_tag>

-e DB_URI=<your_db_uri> here -e sets the environment variable. I set my DB with URI, so I passed this environment variable DB_URI while running my image.

This is it! 🥳 If you have any questions or face any problem, comment below 👇

How to add some badges in your git readme (GitHub, Gitlab etc.)

Azizul Haque Ananto — Thu, 04 Jun 2020 21:34:22 +0000

While I published this repo I was looking for some badges to add in my readme. And found some interesting services worth sharing.

The most amazing place to make your badges is the https://shields.io. But I will explain each category to use them in the shield.

Build

First, the badge you will look for is the build passing badge. You have written some amazing tests and you need to show that they are passing. There are several good services for that, I used Travis CI.

The integration is really easy. You need to add a .travis.yml file in your repo. Mine was pretty straight-forward -

language: python
python:
  - "3.7"
install:
  - pip install -r requirements.txt
script: 
  - python -m pytest  --cov-report term --cov=app
after_success:
  - codecov

I will talk about the after_success part next. Before that, you need the badge, right? You can get it by simply clicking the build passing button.

Code Coverage

You have the tests passing, and the earned the badge. Now you need some coverage for your tests. You already know it by running tests with coverage option. For Python you may run this python -m pytest --cov-report term --cov=app. But how to get the badge?

You can use Codecov. For Python you need to install the codecov package by pip install codecov. Then just add this line in your .travis.yml -

after_success:
  - codecov

You can find the badge in the settings page.

You can also use Coveralls for the same purpose. In that case, you need to install coveralls package in Python. You can find the badge in the bottom of your repo page in coveralls.

Dependency monitor

This is really useful, not only for the badge but you can also manage your dependencies with the coming updates.

For Python you can use requires.io. To get PR for new package changes you need to configure it by these steps: Hooks -> New pull request -> (configure as you like) -> save changes. And you can find the badge in specific repo page.

For Node.js Depfu is a good service. You will get PR automatically, don't need to setup any hooks. You can find the badge in the settings of the specific repo page.

Code Quality

There's a few services for that. I used both lgtm and Code Climate. I won't comapre or judge them as they both marked my code A 😅

You can find lgtm badges in the specific repo's integration page and the codeclimate badges can be found in the repo settings page.

There are also other services specific to your needs. You can know about them from shields actually. Making badge is really easy with them. Have a try!

Why a distributed system (highly concurrent) is mostly asynchronous?

Azizul Haque Ananto — Thu, 26 Mar 2020 07:51:31 +0000

The actual problem in synchronous systems is the waiting time between the client and server where both have to wait for the process to complete. In a system where millions of requests are made per second, they usually take the request and process it asynchronously with messaging systems like Kafka or any MQ's. A gateway client may wait for the response event to respond to the user, but the actual process happens in async (maybe) in the internally distributed systems. So there are two questions I would like to ask -
1) Is my concept correct about high concurrent systems?
2) Is there any way we can make this kind of system synchronous? (internally)