Creating N1QL Labelled Image Database using Couchbase, FastAPI, and SerpApi
Tags: machinelearning, fastapi, couchbase, serpapi
Published: 2022-06-16 00:19:05

This is a part of the series of blog posts related to Artificial Intelligence Implementation. If you are interested in the background of the story or how it goes:

|Previous blog post links|
|---|
|[How to Train a Scalable Classifier with FastAPI and SerpApi ?](https://serpapi.com/blog/scalable-classifier-serpapi-fastapi)|
|[Automatic Training using FastAPI, Pytorch and SerpApi](https://serpapi.com/blog/automatic-training-fastapi-pytorch-serpapi)|
|[Creating N1QL Labelled Image Database using Couchbase, FastAPI, and SerpApi](https://serpapi.com/blog/creating-n1ql-labelled-image-database-couchbase-fastapi)|

On the previous weeks we explored how to create your own image dataset using [SerpApi's Google Images Scraper API](https://serpapi.com/images-results) automatically, and used those images to automatically train a network using simple command object to be passed to FastAPI. This week we will improve the database creation method by using Couchbase as a Storage Server, and will show how to fetch a random element from a given subset.

## Couchbase Configuration
We will be needing Couchbase Community server, and Python SDK for Couchbase for this project. You can find relevant information on [Couchbase Docs](https://docs.couchbase.com/server/current/install/install-intro.html).

For this tutorial we will be using CouchBase version 7.1.0 for Debian 11:
![couchbase](https://user-images.githubusercontent.com/73674035/173905887-067a359a-f824-4a0d-b080-7a54b00f654c.png)
You can access it from [this link](https://www.couchbase.com/downloads)

Once you install it, define username and password from the server destination (`http://kagermanov:8091` in my case), you will be greeted with such dashboard:
![server](https://user-images.githubusercontent.com/73674035/173906456-e1e894a7-be72-4daa-b74a-d3741ea56ea7.png)
This means you have successfully deployed the server.
For those of you who want to stop the background process on Linux, you can type `sudo systemctl stop couchbase-server` to stop the server at your will.

Head to `Buckets` on the left hand menu and add a new bucket called `images` from `ADD BUCKET` button:
![add_bucket](https://user-images.githubusercontent.com/73674035/173907054-6f2c7673-1c74-470e-8ebb-d2dff801c7d9.png)
Make sure to choose a ram amount that won't force your local system into frenzy. I

Now, You need to add a scope and collection within this bucket via `Scopes & Collections` button:
![scopes_and_collections](https://user-images.githubusercontent.com/73674035/173915246-f204dbb8-57ea-4f6f-b08b-c26201b367a1.png)

Add a scope named `image`, and within it, a collection named `labelled_image`:
![add_scope](https://user-images.githubusercontent.com/73674035/173915719-4b0afba3-7e0b-4556-91d8-2f16900a1ac7.png)
![add_collection](https://user-images.githubusercontent.com/73674035/173915740-b77929ce-d848-415c-a571-c43ab7d61127.png)

Next, head to playground where you can do a manual query, and run the following:

```sql
CREATE PRIMARY INDEX ON `images`;
```
![manual_query](https://user-images.githubusercontent.com/73674035/173916359-df78a944-9b76-4bd0-bb23-3869e0191c5b.png)

Lastly, make sure you install the [Couchbase Python SDK](https://pypi.org/project/couchbase/) via pip, and everything is set for our server.


## Automatic Image Collector

Let's create a seperate file within our project called `add_couchbase.py`. This will be the refactored version of `add.py` which was automatically gathering images with a certain query.
Here are the requirements for it:
```py
from couchbase.options import (ClusterOptions, ClusterTimeoutOptions, QueryOptions)
from couchbase.auth import PasswordAuthenticator
from couchbase.cluster import Cluster
from serpapi import GoogleSearch
from datetime import timedelta
from pydantic import BaseModel
import mimetypes
import requests
import random
import uuid
import urllib
import base64
```
To break them down for their specific usecase:
|Requirement|Explanation|
|---|---|
|couchbase.options.ClusterOptions|For defining timeout and authentication options of storage cluster connection|
|couchbase.options.ClusterTimeoutOptions|For maximum time allowed to perform an operation in the cluster|
|couchbase.options.QueryOptions|For giving positional parameters within a N1QL query to Couchbase Server|
|couchbase.auth.PasswordAuthenticator|For authenticating connection with the storage server|
|couchbase.cluster.Cluster|For the connection to the storage server|
|serpapi.GoogleSearch|For Scraping links of images from Google Images|
|datetime.timedelta|For defining a time interval|
|pydantic.BaseModel|For defining objects to be passed as parameters|
|mimetypes|For guessing the extension of the image|
|requests|For making GET requests to image links|
|random|For creating a random number within an interval to fetch random image from storage database|
|uuid|For creating a unique identifier for each image|
|base64|For converting image byte data to base64 for storing purposes|

Let's define a pydantic base model for an image to be uploaded to the Couchbase Storage Server:
```py
class Document(BaseModel):
	type: str = "image"
	id: str
	classification: str
	base64: str
	uri: str
```

`id` will be a unique uuid of the image for image to be called manually in the future.
`classification` is the query given to [SerpApi](https://serpapi.com)'s Engine.
`base64` will be the the string representation of the image to be recreated within a training session.
`uri` will represent the url we fetch the image from.

Let's initialize our Storage Database in a class:
```py
class ImagesDataBase:
	def __init__(self):
		username = "<Your Couchbase Username>"
		password = "<Your Couchbase Password>"
		bucket_name = "images"
		auth = PasswordAuthenticator(
				username,
				password
		)
		timeout_opts = ClusterTimeoutOptions(kv_timeout=timedelta(seconds=10))
		self.cluster = Cluster('couchbase://localhost', ClusterOptions(auth, timeout_options=timeout_opts))
		self.cluster.wait_until_ready(timedelta(seconds=5))
		cb = self.cluster.bucket(bucket_name)
		self.cb_coll = cb.scope("image").collection("labelled_image")
```

Here's the function for inserting an image to the Storage server with a unique id:
```py
	def insert_document(self, doc: Document):
		doc = doc.dict()
		print("\nInsert CAS: ")
		try:
			key = doc["type"] + "_" + str(doc["id"])
			result = self.cb_coll.insert(key, doc)
			print(result.cas)
		except Exception as e:
			print(e)
```
`doc` in this context represents the Document object we store the image inside to be uploaded to the Couchbase Server.

Let's have another function to call an image by its unique key we will generate. This function will not be used in the context of this blog post.
```py
	def get_image_by_key(self, key):
		print("\nGet Result: ")
		try:
			result = self.cb_coll.get(key)
			print(result.content_as[str])
		except Exception as e:
			print(e)
```

Next, we need to build a function in which helps us upload only the unique images. The differentiator will be the unique link within the scope of a classsification:
```py
	def check_if_it_exists(self, link, cs):
		try:
			sql_query = 'SELECT uri FROM `images`.image.labelled_image WHERE classification = $1 AND uri = $2'
			row_iter = self.cluster.query(
				sql_query,
				QueryOptions(positional_parameters=[cs, link]))
			for row in row_iter:
				return row
		except Exception as e:
			print(e)
```
This function takes `link`, which is the link to the image, and `cs`, which is the classifier of the image. If an image with the same link does not already exist within our storage, it returns `None`. The reason we don't query the entire database for the uniqueness is simple. First, it wouldn't be efficient in the long run. Second, same images could have different classifications. Imagine the logo of `Apple, the company`. It is also `Apple, the fruit`. If we are classifying between `Apple Logo` and `Blackberry Logo`, and if the image is in `Apple` classification only, there is a chance that the model could fail to interpret. This approach might create unnoticable duplicate images with different classifications But in the long run it would prove useful.
Here's an example of the following manual query that we know already exists in the Couchbase Server:
```sql
SELECT uri FROM `images`.image.labelled_image WHERE classification = 'Pomegrenate' AND uri = 'https://i0.wp.com/post.healthline.com/wp-content/uploads/2022/02/pomegranate-seeds-fruit-1296x728-header.jpg?w=1155&h=1528'
```
![uri_found](https://user-images.githubusercontent.com/73674035/173935876-9ba88709-f9e6-43de-a653-494d90eb7955.png)



Now that we have the uniqueness out of the way, let us focus on randomness of a given query. This part will also not be used in this blog post, but for future purposes. The function will give the number of images in a subset of classifications. It is useful for determening a random number within range of maximum number size.
```py
	def get_max_image_size(self, cs):
		try:
			sql_query = 'SELECT COUNT(*) as max_items FROM `images`.image.labelled_image WHERE classification = $1'
			row_iter = self.cluster.query(
				sql_query,
				QueryOptions(positional_parameters=[cs]))
			for row in row_iter:
				return row
		except Exception as e:
			print(e)
```
Here's another example query for the size of Orange Images in the storage server:
```sql
SELECT COUNT(*) as max_items FROM `images`.image.labelled_image WHERE classification = 'Orange'
```
![orange_count](https://user-images.githubusercontent.com/73674035/173937013-8539f14a-1ebb-4000-a888-551fb1b03cab.png)

Now, let's define a function that randomly picks a number for us. For this we will define a random integer outside the scope of the query we will feed. But we will define this random integer with the previous function we constructed:
```py
	def random_lookup_by_classification(self, cs):
		max_size = self.get_max_image_size(cs)['max_size']
		random_number = random.randint(0,max_size - 1)
		print("\nLookup Result: ")
		try:
			sql_query = 'SELECT (SELECT im.base64 FROM `images`.image.labelled_image AS im WHERE im.classification = $1)[$2]'
			row_iter = self.cluster.query(
				sql_query,
				QueryOptions(positional_parameters=[cs, random_number]))
			for row in row_iter:
				return row
		except Exception as e:
			print(e)
```
Here is an example query with the random number 37, which is between 0 and 103(from 104 images of Orange):
```sql
SELECT (SELECT im.base64 FROM `images`.image.labelled_image AS im WHERE im.classification = 'Orange')[37]
```
![random_found](https://user-images.githubusercontent.com/73674035/173938845-2046ec59-1ba3-4f24-bdd8-acc8bc8467c0.png)

We have eveything we need for this week and the coming week's blog post now. Let's redefine what we have already defined. A pydantic model for the `Query` object we pass to the endpoint:
```py
class Query(BaseModel):
		google_domain: str = "google.com"
		num: str = "100"
		ijn: str = "0"
		q: str
		api_key: str ## You may replace this with `api_key: str = "Your API Key"`
```
Again, the API key mentioned here is your unique API key for [SerpApi](https://serpapi.com). It can be accessed via [Api Key page](https://serpapi.com/manage-api-key).

Here's the redefinition of the `Download` Class. We can omit some parts to keep uniqueness, and add new ones like database object.
```py
class Download:
	def __init__(self, query: Query, db: ImagesDataBase):
		self.db = db
		self.query = query
		self.results = []
```

There is no change in the fucntion of [SerpApi's Google Images API](https://serpapi.com/images-results) implementation. However, let me restate one amazing fact again. If the query you are searching is cached, you can get it free of charge.
```py
	def serpapi_search(self):
		params = {
			"engine": "google",
			"ijn": self.query.ijn,
			"q": self.query.q,
			"google_domain": self.query.google_domain,
			"tbm": "isch",
			"num": self.query.num,
			"api_key": self.query.api_key
		}

		search = GoogleSearch(params)
		results = search.get_dict()
		results = results['images_results']
		self.results = results = [x['original'] for x in results]
```

Let's define another function for downloading and image and returning it as a Document object:
```py
	def get_document(self, link):
		print("Downloading {}".format(link))
		classification = self.query.q
		r = requests.get(link)
		base64_str = base64.b64encode(r.content).decode('ascii')
		extension = mimetypes.guess_extension(r.headers.get('content-type', '').split(';')[0])
		id = uuid.uuid1().hex
		if extension == ".jpg" or extension == ".jpeg" or extension == ".png":
			doc = Document(id = id, classification = classification, base64 = base64_str, uri = link )
			return doc
		else:
			return None
```

Next, we define the function to insert the Document objects we get from the previous function. We check for the uniqueness of the link to reduce duplicates in this function also:
```py
	def move_to_db(self, link):
		doc = self.get_document(link)
		sameness = self.db.check_if_it_exists(self.query.q, link)
		if doc is not None and sameness is None:
			self.db.insert_document(doc=doc)
```

Here, we can iterate through all the links gathered from [SerpApi's Google Images Scraper API](https://serpapi.com/images-results), and upload them to our Couchbase Storage Server:
```py
	def move_all_images_to_db(self):
		self.serpapi_search()
		for result in self.results:
			try: 
				self.move_to_db(result)
			except:
				"\n Passed image"
```

Now that we have everything in place, let us define the `add_couchbase.py` function within our `main.py`:
```py
from fastapi import FastAPI
from add_couchbase import Download, Query, ImagesDataBase
from create import CSVCreator, ClassificationsArray
from dataset import CustomImageDataLoader, CustomImageDataset
from train import CNN, Train
from commands import TrainCommands

app = FastAPI()

@app.get("/")
def read_root():
  return {"Hello": "World"}

@app.post("/add_to_db/")
def create_query(query: Query):
  db = ImagesDataBase()
  serpapi = Download(query, db)
  serpapi.serpapi_search()
  serpapi.move_all_images_to_db()
  return  {"status": "Complete"}

...
```

## Collecting Images and Storing Them with Classifications

Let's put everything we made into practice. Run the server with the following command:
```shell
uvicorn main:app --host 0.0.0.0 --port 8000
```
and then head to `localhost:8000/docs` to try out `/add_to_db/` endpoint with the following request body:
```py
{
  "google_domain": "google.com",
  "num": "100",
  "ijn": "0",
  "q": "string",
  "api_key": "<Your API Key>"
}
```
![post_add](https://user-images.githubusercontent.com/73674035/173942534-c56f8077-209c-4641-855c-344089745d64.png)

If you observe the terminal, you will see that the process of updating the database is happening in real time:
```shell
...
Insert CAS: 
1655331991397793792
Downloading https://target.scene7.com/is/image/Target/GUEST_c3800365-97f3-4fe9-8061-8894a378cc85?wid=488&hei=488&fmt=pjpeg

Insert CAS: 
1655331991890821120
Downloading https://solidstarts.com/wp-content/uploads/Mango_edited-scaled.jpg

Insert CAS: 
1655331993304432640
Downloading https://www.netmeds.com/images/cms/wysiwyg/blog/2019/04/Raw_Mango_898.jpg
...
```

If we query the database even before it finished, we can see that the entries with the classification label `Mango` are being updated. Here's the command for it:
```sql
SELECT COUNT(*) as max_items FROM `images`.image.labelled_image WHERE classification = 'Mango'
```
![query_count_2](https://user-images.githubusercontent.com/73674035/173942931-96f39cb1-2b8d-4b57-9835-ccf2f908822f.png)
It already added 67 unique images to the database we can use to train our network in the coming weeks.

## Conclusion

N1QL Databases such as Couchbase have fast response times compared to other Storage Databases. In this regard, I thought refactoring this part as an essential step before takign any further actions. This implementation will provide us with the speed and scalability we hope to support us in the coming week's challanges in comparing different approaches in Image Classification. It is also important for async handling of some functions such as inserting Images instead of naming them using the OS. 

I am grateful the user for their attention, and all the support of [Brilliant People of SerpApi](https://serpapi.com/team). In the coming weeks, we will explore utilizing async handling, and corotines to lower the response time of the actions. If you are interested in the tutorials just like these, feel free to sign up to [SerpApi Blog](https://serpapi.com/blog/#/portal/signup), or follow us on the medium you have found us. Your support is the primary factor in making such tutorials.