FastAPI, PyMongo and ‘ObjectId’ object is not iterable.

If you’re working with FastAPI and PyMongo, it is very likely that you are greeted with:

ValueError: [TypeError("'ObjectId' object is not iterable"), TypeError('vars() argument must have __dict__ attribute')]

As you may know, every document in MongoDB has an _id attribute (a primary key). When you retrieve a document with PyMongo and you return the dictionary, FastAPI cannot serialize _id (which is of type bson.objectid.ObjectId). If you look around there are a number of solutions to this. Check them out, as they might be more appropriate for your case. However, one particular workaround, namely doing a document.pop('_id', None) before returning the dictionary got me thinking. So for my particular case I tried:

:
id = bucket_list.insert_one(document).inserted_id
# document.pop('_id', None)
document["_id"] = f'{id}'
:

which worked find for my use-case.

Return a blank favicon.ico with Python bottle

Bottle is a fine framework when you need to quickly start a small Python application and serve web calls from it. However, you will soon notice that browser calls to your application ask for favicon.ico. Of course you can ignore those calls, but what if you want to serve them too?

Once choice is to use server_static() and return the icon as described here. But what if you want to return a blank icon, since this is a quick hack anyway and you do not want to spend time searching for whatever icon or not?

Starting from the description of a blank favicon, you can add this to your code:

@get('/favicon.ico')
def get_favicon():

    response.content_type = 'image/x-icon'

    return "data:image/x-icon;base64,iVBORw0KGgoAAAANSUhEUgAAABAAAAAQEAYAAABPYyMiAAAABmJLR0T///////8JWPfcAAAACXBIWXMAAABIAAAASABGyWs+AAAAF0lEQVRIx2NgGAWjYBSMglEwCkbBSAcACBAAAeaR9cIAAAAASUVORK5CYII="

Mass disabling all Jenkins jobs

There are times that you need to disable all jobs on a Jenkins server. Especially when you’ve made a backup copy for testing or other purposes. You do not want jobs to start executing from that second server before you’re ready. Sure you can start Jenkins in quiet mode but sometime you have to exit it and scheduled jobs will start running. What can you do?

Well, there are plenty of pages that show Groovy code that allows you to stop jobs, and there are even suggestions to locate and change every config.xml file by running something like sed -i 's/disabled>false/disabled>true/' config.xml on each of them. Or even better use the Configuration Slicing plugin. Firstly, you may feel uneasy to mass change all config.xml file from a process external to Jenkins. Secondly, the Configuration Slicing plugin does not give you a "select all option" nor does it handle Multibranch Pipeline jobs. Thirdly, the Groovy scripts I’ve found shared by others online, also do not handle Pipelines and Multibranch Pipelines. If you’re based on Multibranch Pipelines, you’re kind of stuck then. Or you have to go and manually disable each one of them.

Thankfully there’s a solution using Jenkins’s REST API and python-jenkins. An example follows:

import jenkins

server = jenkins.Jenkins('http://127.0.0.1:8080/',
        timeout=3600,
        username=USERNAME,
        password=PASSWORD)


all_jobs = server.get_all_jobs()
for j in range(len(all_jobs)):
    try:
        server.disable_job(all_jobs[j]['fullname'])
    except Exception as e:
        print(all_jobs[j]['fullname'])

I hope it helps you out maintaining your Jenkins.

How to get all the indices in ElasticSearch with Python?

This seems to be a pretty common question. And the most common answer is to use the Python ElasticSearch client and the get_alias() method like this:

import Elasticsearch

es = elasticsearch.Elasticsearch(hosts=[ES_HOST], )
idx_list = [x for x in es.indices.get_alias("*").keys() ]

This is the most common answer one can see in StackOverflow. But ElasticSearch offers us the cat API which is better suited for such a query. So a better way to approach this can be:

import Elasticsearch

es = elasticsearch.Elasticsearch(hosts=[ES_HOST], )
idx_list = es.cat.indices(index='foobar-20*', h='index', s='index:desc').split()

The above example asks an even more elaborate query: Of all the indices, return to us those who match the pattern foobar-20*, return only the index name from the fields that the cat API returns, and by the way, sort the returned index names in descending order.

If the database offers us a way to do things, it is best that we ask it to.

fizzbuzz

Years ago, while browsing the original wiki site, I stumbled upon the fizzbuzz test:

“Write a program that prints the numbers from 1 to 100. But for multiples of three print “Fizz” instead of the number and for the multiples of five print “Buzz”. For numbers which are multiples of both three and five print “FizzBuzz”.”

Imagine my surprise when reading that it was designed to make most programming candidates fail the test. From time to time I have the fine opportunity to introduce people who have never coded before in their lives to Python. Within the first two hours they have managed to produce a fizzbuzz version that looks like this:

for i in range(1, 101):
  if i % 3 == 0 and i % 5 == 0:
    print("fizzbuzz")
  elif i % 3 == 0:
    print("fizz")
  elif i % 5 == 0:
    print("buzz")
  else:
    print(i)

I like the myth that 99.5% of candidates fail fizzbuzz. I tell students that they now can celebrate their achievement. But this is not where I stop with the test. I can now tell them about functions and then have them make one like:

def fizzbuzz(start, stop):
  :

where I have them modify their code above to make it a function. And afterwards they can learn about named parameters with default values:

def fizzbuzz(start=1, stop=100, three=3, five=5):
  :

Notice above how one can change the values of 3 and 5 from the original test and try a different pair. And yes, I leave the functions as exercises to the reader :)

But the best sparks in their eyes come when they remember that I had taught them certain properties of strings some three hours ago, like:

>>> 'fizz' + 'buzz'
'fizzbuzz'
>>> 'fizz' * 1
'fizz'
>>> 'fizz' * True
'fizz'
>>> 'fizz' * 0
''
>>> 'fizz' * False
''
>>> 'fizz' + ''
'fizz'
>>>

So their first observation that 'fizz' + 'buzz' equals 'fizzbuzz' is followed by what is summed up by the table:

String Sum of strings Expanded sum
‘fizzbuzz’ ‘fizz’ + ‘buzz’ ‘fizz’ * True + ‘buzz’ * True
‘fizz’ ‘fizz’ + ” ‘fizz’ * True + ‘buzz’ * False
‘buzz’ ” + ‘fizz’ ‘fizz’ * False + ‘buzz’ * True
” + ” ‘fizz’ * False + ‘buzz’ * False

Which makes them write something like that in the end:

def fizzbuzz(x, y):
  return 'fizz' * x + 'buzz' * y

for i in range(1, 101):
  print(fizzbuzz(i % 3 == 0, i % 5 == 0) or i)

How many things one can learn within a day starting from zero using the humble (and sometimes humbling) fizzbuzz.

Let’s do a Koch snowflake

Good friend Dimitris, after reading my previous post pointed me to Koch snowflakes. How cool is a line of infinite length that covers a finite surface! A Koch snowflake turns out to be easily constructed with turtle as suggested by the Wikipedia article. Well, you also get to learn about Thue-Morse sequences and evil numbers in the process. To be honest, this is also a good toy case, using a real sequence, to learn how to use yield.

Koch snowflake
Koch snowflake

import turtle
import functools

# Compute the next digit of the Thue-Morse sequence
# https://oeis.org/A010060
# Learn about evil and oddium numbers in the process.

def thue_morse_seq(n=0):
  while True:
    yield functools.reduce(lambda x, y: x + y, map(int, bin(n)[2:])) % 2
    n += 1

if __name__ == "__main__":

  window = turtle.Screen()
  window.bgcolor('light gray')

  pen = turtle.Turtle()
  pen.speed(20)
  pen.color('dark blue')
  pen.pensize(1)
  pen.shape('classic')
  pen.penup()
  pen.setpos(0, 0)
  pen.pendown()

  n = thue_morse_seq(0)
  while True:
    if next(n) == 0:
      pen.forward(2)
    else:
      pen.left(60)

[pastebin]

Rule 110 with Turtle

I like repl.it a lot and from time to time I use it for 10 liners instead of my command line. Especially when I do not want to install a brand new language implementation or even creating a new Python environment for five minutes.

I was thinking about Rule 110 and how most of the examples I’ve seen from hobbyists are ASCII based as opposed to more proper, nicer graphics by researchers of the CA area. And I was thinking whether using Turtle could be used to display it a bit better in one’s spare time. That is because I still have not figured out how TkInter works with repl.it. It turns out that you can make something nice with Turtle:

rule 110
Rule 110 elementary cellular automaton

I have pasted the rather rudimentary, repl.it ready, python code here.

unbound, python and conditional replies based on source IP address

We’re using unbound internally for DNS resolution. It works smoothly and allows for some DNS tricks when you want to implement some split-brain trickery, but not a complete split-brain deployment.  The other day we needed to send out conditional replies based on the IP address of the querying machine.  Unbound comes with a python module but it has some of the weirdest, unhelpful documentation ever.  I am not alone in believing this.

It is very hard to figure out the source IP address of a DNS query using the unbound python library. My first pointer on how to do so was on ServerFault.  I have uploaded my own version of an operate function at pastebin. The code in question that you need to consider is:

# Find out source IP address
rl = qstate.mesh_info.reply_list
while (rl):
  if rl.query_reply:
    q = rl.query_reply
    break
  rl = rl.next

# Careful with this conditional
try: addr = q.addr
except NameError: addr = None

The try … except handling is needed because I found out that sometimes the q.addr may not be defined and thus further down the line you may be bitten by an abnormal exit of your script.

Update: two friends have suggested that I change the while loop with a more Pythonic list comprehension:

q = next((x for x in qstate.mesh_info.reply_list if x.query_reply), None)
try: addr = q.query_reply.addr
except NameError: addr = None

One of them actually has a pretty cool pastebin about it.