Artetore Imagehub is a hub to store and showcase personalized image

The backend of the Image Hub is developed by FastAPI.

All requirements of the backend are:

fastapi
python-jose[cryptography]
passlib[bcrypt]
SQLAlchemy
python-dotenv
uvicorn
mysql-connector-python

The initial step to creating the backend is to develop some routing addresses for the API. But not all routing address is easy to set up. So I started with login and signup router. But here I didn't set up any signup router because initially, I thought there would be only one and only admin for the image hub. So if I add an endpoint of signup routing, it would make a security risk. Hackers may try to create an admin account for them. But a signup routing can be integrated anytime. So after adding the username and password as hardcoded input directly into the database, I started working on the login routing endpoint

The login endpoint function takes two inputs from the client and one function of the database that will be added as a Dependency parameter and this dependency is known as Dependency Injection. In Fastapi "Dependency Injection" means, there is a way for our code to declare things that it requires to work and use. After getting the username and password value from the client through the login function, I need to validate it, and to do that I sent the username and password to a different function. That function name is verify_login and the function returns two values. One is the result and the other is new_token. The result is an integer value that represents some predefined status of operation and the new_token valu contains JSON Web Token. This backend session is protected by JWT so there is very little chance of bypassing the system by hackers and doing some manipulation in the database.

Now if depending on the result value, the function will send response different things to the client. For example,

result == 0: username or password did not match [user doesn't exist]
result == 1: username or password did not match [password doesn't match]
result == 2: new_token

The reason behind sending the same response for the user does;t exist and the credentials don't match is, I don't want the user to know that the user exists or he is giving the wrong password. Because If I make it specific, hackers may try to do some funny things by trying brute-forcing or dictionary attacks. But the different responses can be generated anytime if I want it. The development is done in a flexible way

from fastapi import APIRouter, Depends, status
from starlette.responses import JSONResponse
from router.image import get_db
from sqlalchemy.orm import Session
from logic.admin import verify_login

router = APIRouter(
    prefix="/login",
    tags=["login"]
)


@router.get("/")
async def login(username: str, password: str, db: Session = Depends(get_db)):
    result, new_token = verify_login(db, username, password)
    if result == 2:
        return JSONResponse(
            status_code=status.HTTP_200_OK,
            content={
                "code": 2,
                "token": new_token
            }
        )
    elif result == 0:
        return JSONResponse(
            status_code=status.HTTP_200_OK,
            content={
                "code": 0,
                "hint": "username or password did not match"
            }
        )

    return JSONResponse(
        status_code=status.HTTP_200_OK,
        content={
            "code": 1,
            "hint": "username or password did not match"
        }
    )

This project is all dependent on if one drive works seamlessly with my backend or not. So the second phase of the development was to connect with one drive and make an endpoint for it on fast API.

So I created a routing endpoint named admin. This routing file will contain all the necessary endpoints for doing admin operations. The endpoints that are added in that file are:

onedrive_connection: Accepts JWT token as input and returns if one drive connection is successful or not

  @router.get("/onedrive/connect")
  async def onedrive_connection(token: str = Depends(oauth2_bearer), db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      connection_dict = connect_onedrive(db, user_id)
      if connection_dict["connected"] == 1:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 1,
                  "hint": "connection_successful"
              }
          )
      url = generate_onedrive_auth_url()
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 0,
              "hint": "connection_failed",
              "url": url
          }
      )

onedrive_folders: Accepts JWT token as input and returns folder list of one drive folder list. The root folder id is added to the .env file and the key for the root folder is folder_arts_id

  @router.get("/onedrive/folders")
  async def onedrive_folders(token: str = Depends(oauth2_bearer), db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      token_status, folders = get_folders(db, user_id)
      if token_status == 0:
          url = generate_onedrive_auth_url()
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 0,
                  "hint": "refresh_token_expired_need_manual_authorization",
                  "url": url
              }
          )
      if token_status == 2:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 99,
                  "hint": "data_extraction_failed_from_json"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 1,
              "folders": folders
          }
      )

add_folder_operation: Takes input of folder_name, JWT and returns the result of operation

  @router.get("/onedrive/add-folder")
  async def add_folder_operation(folder_name: str, token: str = Depends(oauth2_bearer), db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      result = create_folder_in_onedrive(db, user_id, folder_name)
      if result == 1:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 1,
                  "hint": "folder_creation_operation_successful"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 0,
              "hint": "folder_creation_operation_failed"
          }
      )

rename_operation: Takes input of item_id, new_name, JWT and returns result of operation

  @router.get("/onedrive/rename")
  async def rename_operation(item_id: str, new_name: str, token: str = Depends(oauth2_bearer),
                             db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      result = rename_onedrive_item(db, user_id, item_id, new_name)

      if result == 1:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 1,
                  "hint": "rename_operation_successful"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 0,
              "hint": "rename_operation_failed"
          }
      )

folder_delete_operation: Takes input of folder_id, JWT and returns the result of the operation. This can also work as a file deletion operation

  @router.get("/onedrive/delete-folder")
  async def folder_delete_operation(folder_id: str, token: str = Depends(oauth2_bearer), db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      result = delete_folder_from_onedrive(db, user_id, folder_id)
      if result == 1:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 1,
                  "hint": "folder_deletion_operation_successful"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 0,
              "hint": "folder_deletion_operation_failed"
          }
      )

file_upload_session: Takes input of folder_id, file_name, and JWT and returns one drive session URL to upload on that specific folder

  @router.get("/onedrive/upload-session")
  async def file_upload_session(folder_id: str, file_name: str, token: str = Depends(oauth2_bearer),
                                db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      success_status, upload_url = get_onedrive_uploading_session(db, user_id, folder_id, file_name)
      if success_status == 1:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 1,
                  "sessionUrl": upload_url,
                  "hint": "got_session"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 0,
              "hint": "couldn't_get_session"
          }
      )

groups: Returns only group list from database [groups are used to add multiple images from different angles and positions in the same group]

  @router.get("/groups")
  async def group_list(token: str = Depends(oauth2_bearer), db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      groups = get_group_list(db)
      return {
          "data": groups
      }

add_image_in_database_operation: Takes image_list, JWT token as input and returns operation status

The input_list will be added as below:

  {
  "image_list":{
      "01TX4ZA735SY5SYKPEY5HJ7XW6CC2FUBRY": {
          "all_image_id": [
              {
                  "file_name": "steps-in-instruction-execution-by-cpu.png",
                  "file_id": "01TX4ZA77HMAJQR72NJZC2ZNJEZEK22ILO"
                  }
              ]
              ,
              "group": -1,
              "folder_name": "System"
          }
      }
  }

The function of the endpoint is:

  @router.post("/add/image")
  async def add_image_in_database_operation(image_list: AddImageList, token: str = Depends(oauth2_bearer),
                                            db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      result = add_image_in_database(db, user_id, image_list.image_list)
      if result == 1:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 1,
                  "hint": "image_added_in_database"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 2,
              "hint": "invalid image list"
          }
      )

delete_image_in_database_operation: Takes image_list, JWT token as input and returns operation status

image_list will be defined as below:

  {
      "image_list":{
          "17": [
              {
                  "folder_name": "System",
                  "image_id": 1,
                  "image_name":"steps-in-instruction-execution-by-cpu.png"
              }
          ]
      }
  }

The function endpoint to process the data is given below:

  @router.post("/delete/image")
  async def delete_image_in_database_operation(image_list: AddImageList, token: str = Depends(oauth2_bearer),
                                               db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      result = delete_image(db, image_list.image_list)
      if result == 1:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 1,
                  "hint": "image_deleted_from_database"
              }
          )

get_single_image_from_group: Takes group_id, image_id, JWT token and returns low-resolution image


  @router.get("/image")
  async def get_single_image_from_group(group_id: int, image_id: Union[int, None] = None,
                                        token: str = Depends(oauth2_bearer),
                                        db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      code, image = get_one_low_res_image_from_group(db, group_id, image_id)
      if code == 1:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 1,
                  "image": image,
                  "hint": "got_image"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 0,
              "hint": "couldn't_get_image"
          }
      )

get_groups_with_images: Takes JWT token as input and returns all group list with images

  @router.get("/groups/images")
  async def get_groups_with_images(token: str = Depends(oauth2_bearer),
                                   db: Session = Depends(get_db)):
      user_id = get_current_user_from_jwt_token(token)
      result = get_group_list_with_images(db)
      if result == {}:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 0,
                  "hint": "empty_database"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 1,
              "data": result
          }
      )

Now the endpoint will only accept the client request through the endpoint but the data will be processed in some other functions. Those functions are added in the Python files of logic folders.

At first, I need to make logic for the login routing endpoints functions. The login uses the verify_login function and that function uses a function from the cryptography.py file. In that file there are two functions:

The first function is to get a password hash from a normal string password

  def get_password_hash(password: str):
      return bcrypt_context.hash(password)

The second function is to validate the normal string password nad hash password to check if both are the same

  def verify_password(password: str, hashed_password: str):
      return bcrypt_context.verify(password, hashed_password)

Then I had to make logic for JWT. And I added all the functions inside the token.py file. In that file, there are two functions.

The first function generates a JWT token and returns the token

  def create_jwt_access_token(user_id: str, expires_delta: Optional[timedelta]):
      encode = {
          "sub": "Artetore Token",
          "id": user_id,
      }
      if expires_delta:
          expire = datetime.utcnow() + expires_delta
      else:
          expire = datetime.utcnow() + timedelta(minutes=60)
      encode["exp"] = expire
      return jwt.encode(encode, SECRET_KEY, algorithm=ALGORITHM)

The second function is used to validate the JWT token

  def get_current_user_from_jwt_token(token: str = Depends(oauth2_bearer)):
      try:
          payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
          username: str = payload.get("sub")
          user_id: str = payload.get("id")
          if username is None or user_id is None:
              raise HTTPException(
                  status_code=status.HTTP_401_UNAUTHORIZED,
                  detail="Invalid Bearer 1 Token",
              )
          return user_id
      except JWTError:
          raise HTTPException(
              status_code=status.HTTP_401_UNAUTHORIZED,
              detail="Invalid Bearer 2 Token",
          )

Now the complex part comes. Where all the logic contains to manipulate database image and one drive data. All the functions are stored in the admin.py file

The uses of all the functions for one drive data manipulation in that Python script are given below:

generate_onedrive_auth_url: Generates one drive authentication url using the client id and client secret

  def generate_onedrive_auth_url():
      return f"https://login.microsoftonline.com/common/oauth2/v2.0/authorize?client_id={client_id}&scope={scope}&response_type=code&redirect_uri={redirect_uri}"

process_token_from_response_data_and_add_in_db: Takes response data that has the refresh token and current token, user_id and add it to the database

  def process_token_from_response_data_and_add_in_db(response_data: str, db: Session, user_id: str):
      found_tokens = True
      current_token, refresh_token = None, None
      try:
          current_token = json.loads(response_data)["access_token"]
          refresh_token = json.loads(response_data)["refresh_token"]
      except Exception:
          found_tokens = False

      if found_tokens:
          add_tokens_in_database(db, current_token, refresh_token, user_id)
          local_tokens["current_token"] = current_token
          local_tokens["refresh_token"] = refresh_token
      return found_tokens

get_tokens_from_auth_code: Takes auth_code, user_id as input and generate current_token and refresh token

  def get_tokens_from_auth_code(db: Session, auth_code: str, user_id: str):
      token_request_url = 'https://login.microsoftonline.com/common/oauth2/v2.0/token'
      response = requests.post(token_request_url, data={
          "client_id": client_id,
          "code": auth_code,
          "redirect_uri": redirect_uri,
          "client_secret": client_secret,
          "grant_type": "authorization_code"
      })
      return process_token_from_response_data_and_add_in_db(response.text, db, user_id)

verify_login: Takes username and password as input and validate

  def verify_login(db: Session, username: str, password: str):
      hashed_password, user_id = get_hashed_password_from_username(db, username)
      if hashed_password is None:
          return 0, None
      if verify_password(password, hashed_password):
          new_token = create_jwt_access_token(user_id, None)
          return 2, new_token
      return 1, None

get_new_token_from_refresh_token: Takes refresh_token as input and generate new current_token and refresh_token

  def get_new_token_from_refresh_token(refresh_token, db: Session, user_id: str):
      token_request_url = 'https://login.microsoftonline.com/common/oauth2/v2.0/token'
      response = requests.post(token_request_url, data={
          "client_id": client_id,
          "refresh_token": refresh_token,
          "redirect_uri": redirect_uri,
          "client_secret": client_secret,
          "grant_type": "refresh_token"
      })
      return process_token_from_response_data_and_add_in_db(response.text, db, user_id)

get_token_from_db: Takes user_id as input and get current_token and refresh_token from the database and add it to local_token_dict. This local_token_dict is a dictionary that is generated on RAM and is used to manage session data.

  def get_token_from_db(db: Session, user_id: str):
      if local_tokens["current_token"] is None:
          current_token, refresh_token = get_tokens_from_user_id(user_id, db)
          if current_token is not None:
              local_tokens["current_token"] = current_token
              local_tokens["refresh_token"] = refresh_token
          else:
              raise HTTPException(
                  status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                  detail="database_not_found",
              )

connect_onedrive: Connects with one drive and returns the operation result

  def connect_onedrive(db: Session, user_id: str):
      get_token_from_db(db, user_id)

      response = requests.get("https://graph.microsoft.com/v1.0/",
                              headers={'Authorization': 'Bearer ' + local_tokens["current_token"]})
      response_status_code = response.status_code
      connection_onedrive_dict = {
          "connected": 0  # Connection Failed
      }

      if response_status_code == 200:
          connection_onedrive_dict["connected"] = 1  # Connection Successful
      elif response_status_code == 401:
          found_tokens = get_new_token_from_refresh_token(local_tokens["refresh_token"], db, user_id)
          if found_tokens:
              connection_onedrive_dict["connected"] = 1  # Refreshed Token Connection Successful

      return connection_onedrive_dict

get_files_from_folder: Takes one drive folder_id as input and gets file_list of that folder

  def get_files_from_folder(item_id, db, user_id):
      response = requests.get(f"https://graph.microsoft.com/v1.0/me/drive/items/{item_id}/children",
                              headers={'Authorization': 'Bearer ' + local_tokens["current_token"]})
      files_dict = {}
      if response.status_code == 200:
          try:
              response_data = json.loads(response.text)
              for file in response_data["value"]:
                  if file.get("folder") is not None:
                      continue
                  files_dict[file["id"]] = {
                      "file_name": file["name"],
                      "file_size": file["size"],
                      "creation_time": file["fileSystemInfo"]["createdDateTime"],
                      "modification_time": file["fileSystemInfo"]["lastModifiedDateTime"],
                      "is_folder": False
                  }
              return 1, files_dict  # 1 Is For Success
          except Exception:
              print("Exception Found 2")
              return 2, {}
      elif response.status_code == 401:
          found_tokens = get_new_token_from_refresh_token(local_tokens["refresh_token"], db, user_id)
          if found_tokens:
              return get_files_from_folder(item_id, db, user_id)
      return 0, files_dict  # 0 Is For Failure.  Need New Current Token By Manual Authorization

get_folders: Scan arts_folder_id [ From .env ] and return all folders and file list of all children folders

  def get_folders(db: Session, user_id):
      get_token_from_db(db, user_id)
      folders_info_dict = {}
      try:
          response = requests.get('https://graph.microsoft.com/v1.0/me/drive/', headers={'Authorization': 'Bearer ' + local_tokens["current_token"]})
          if response.status_code == 401:
              get_token_from_db(db, user_id)
              return get_folders(db, user_id)
          response_data = json.loads(response.text)
          used = round(response_data['quota']['used'] / (1024 * 1024 * 1024), 2)
          total = round(response_data['quota']['total'] / (1024 * 1024 * 1024), 2)
          folders_info_dict["used_storage"] = used
          folders_info_dict["total_storage"] = total
      except Exception:
          print("Storage Info Fetching Exception")
          return 0, {}
      try:
          response = requests.get(
              f"https://graph.microsoft.com/v1.0/me/drive/items/{folder_arts_id}/children",
              headers={'Authorization': 'Bearer ' + local_tokens["current_token"]})
      except Exception:
          print("Couldn't Connect To Microsoft APi")
          return 0, {}

      if response.status_code == 200:
          print("Response From Folder 1 200")
          try:
              response_data = json.loads(response.text)
              for item in response_data["value"]:
                  if item.get("file") is not None:
                      continue
                  valid_current_token, files_dict = get_files_from_folder(item["id"], db, user_id)
                  folders_info_dict[item["id"]] = {
                      "folder_name": item["name"],
                      "folder_size": item["size"],
                      "creation_time": item["fileSystemInfo"]["createdDateTime"],
                      "modification_time": item["fileSystemInfo"]["lastModifiedDateTime"],
                      "is_folder": True,
                      "files": files_dict.copy()
                  }
                  if valid_current_token == 0:
                      return 0, {}
                  if valid_current_token == 2:
                      return 2, {}
              return 1, folders_info_dict
          except Exception:
              print("Exception Found 1")
              return 2, {}
      elif response.status_code == 401:
          print("Response From Folder 2 Unknown")
          found_tokens = get_new_token_from_refresh_token(local_tokens["refresh_token"], db, user_id)
          if found_tokens:
              return get_folders(db, user_id)
      print("Response From Folder 3 Unknown")
      return 0, folders_info_dict  # 0 Is For Failure.  Need New Current Token By Manual Authorization

create_folder_in_onedrive: Takes folder_name as input and creates a new folder in one drive arts_folder_id and that will be used as the root folder id

  def create_folder_in_onedrive(db: Session, user_id, folder_name: str):
      get_token_from_db(db, user_id)
      body = {
          "name": folder_name,
          "folder": {},
          "@microsoft.graph.conflictBehavior": "rename"
      }
      response = requests.post(f"https://graph.microsoft.com/v1.0/me/drive/items/{folder_arts_id}"
                               "/children", headers={'Authorization': 'Bearer ' + local_tokens["current_token"]},
                               json=body)

      if response.status_code == 201:
          return 1
      elif response.status_code == 401:
          found_tokens = get_new_token_from_refresh_token(local_tokens["refresh_token"], db, user_id)
          if found_tokens:
              return create_folder_in_onedrive(db, user_id, folder_name)
      return 0

rename_onedrive_item: Takes item_id, new_name as input and renames the file/folder name

  def rename_onedrive_item(db: Session, user_id: str, item_id, new_name):
      get_token_from_db(db, user_id)
      response = requests.patch(f"https://graph.microsoft.com/v1.0/me/drive/items/{item_id}",
                                headers={'Authorization': 'Bearer ' + local_tokens["current_token"]},
                                json={"name": new_name})
      if response.status_code == 200:
          return 1
      elif response.status_code == 401:
          found_tokens = get_new_token_from_refresh_token(local_tokens["refresh_token"], db, user_id)
          if found_tokens:
              return rename_onedrive_item(db, user_id, item_id, new_name)
      return 0

delete_folder_from_onedrive: Takes folder_name as input and deletes folder from one drive arts_folder_id

  def delete_folder_from_onedrive(db: Session, user_id: str, folder_id):
      get_token_from_db(db, user_id)
      response = requests.delete(f"https://graph.microsoft.com/v1.0/me/drive/items/{folder_id}",
                                 headers={'Authorization': 'Bearer ' + local_tokens["current_token"]})
      if response.status_code == 204:
          return 1
      elif response.status_code == 401:
          found_tokens = get_new_token_from_refresh_token(local_tokens["refresh_token"], db, user_id)
          if found_tokens:
              return delete_folder_from_onedrive(db, user_id, folder_id)
      return 0

get_onedrive_uploading_session: Takes folder_id, file_name as input and generates one drive session URL and returns it

  def get_onedrive_uploading_session(db: Session, user_id: str, folder_id, file_name):
      get_token_from_db(db, user_id)
      response = requests.post(f"https://graph.microsoft.com/v1.0/me/drive/items/{folder_id}"
                               f":/{file_name}:/createUploadSession",
                               headers={'Authorization': 'Bearer ' + local_tokens["current_token"]})

      if response.status_code == 200:
          upload_url = json.loads(response.text)["uploadUrl"]

          return 1, upload_url
      if response.status_code == 401:
          found_tokens = get_new_token_from_refresh_token(local_tokens["refresh_token"], db, user_id)
          if found_tokens:
              return get_onedrive_uploading_session(db, user_id, folder_id, file_name)
      return 0, None

download_image_content: Takes file_id as input and downloads image on the server from one drive

  def download_image_content(db, user_id, file_id):
      get_token_from_db(db, user_id)
      response = requests.get(f"https://graph.microsoft.com/v1.0/me/drive/items/{file_id}/content",
                              headers={'Authorization': 'Bearer ' + local_tokens["current_token"]})

      if response.status_code == 200:
          return response.content
      return None

The uses of all the functions for database CRUD operation in that Python script are given below:

add_image_in_database: Takes image_list as input and generate image_info_dict and add it to database

  def add_image_in_database(db: Session, user_id: str, image_list: dict):
      found_valid_image = 0
      for folder_key, folder_value in image_list.items():
          all_images = []
          for image_info in folder_value["all_image_id"]:
              image_data = download_image_content(db, user_id, image_info["file_id"])
              if image_data is not None:
                  image_dict = generate_image_dict_for_database(image_data, image_info["file_name"])
                  if image_data is not None:
                      found_valid_image = 1
                      all_images.append(image_dict.copy())
          add_image_info_in_database(db, all_images, folder_value["group"], folder_value["folder_name"], )
      return found_valid_image

generate_image_dict_for_database: Generate image_dict for the database from byte image

  def generate_image_dict_for_database(img_as_byte, file_name):
      low_res_image = change_image_resolution(img_as_byte, 180, 115)
      if low_res_image is None:
          return None
      high_res_image = change_image_resolution(img_as_byte, 700, 525)
      high_res_image_base64 = convert_image_into_base64(high_res_image)
      low_res_image_base64 = convert_image_into_base64(low_res_image)
      full_image = convert_image_into_base64(img_as_byte)
      title = "This Is Sample Title"
      return {
          "title": title,
          "file_name": file_name,
          "low_res_image": low_res_image_base64,
          "high_res_image": high_res_image_base64,
          "full_image": full_image
      }

change_image_resolution: Takes image_data as input and returns custom resolution image

  def change_image_resolution(image_data, target_width, target_height):
      try:
          im = Image.open(io.BytesIO(image_data))
      except Exception:
          return None
      fixed_image = ImageOps.exif_transpose(im)
      im_resize = fixed_image.resize((target_width, target_height))
      buf = io.BytesIO()
      im_resize.save(buf, format=im.format)
      byte_im = buf.getvalue()
      return byte_im

convert_image_into_base64: Convert byte image to base64 image

  def convert_image_into_base64(image_as_byte):
      encoded = base64.b64encode(image_as_byte).decode('ascii')
      base64_image_data = 'data:image/{};base64,{}'.format("png", encoded)
      return base64_image_data

get_one_low_res_image_from_group: Takes group_id, image_id as input and gets the low-resolution image from the database

  def get_one_low_res_image_from_group(db: Session, group_id: int, image_id: Union[int, None]):
      image = get_low_res_image_from_db(db, group_id, image_id)
      if image is None:
          return 0, image
      return 1, image

get_group_list_with_images: Get image_list from the database for each group and returns data

  def get_group_list_with_images(db: Session):
      group_list = get_group_list_with_images_from_db(db)
      group_dict = {}
      for data in group_list:
          if group_dict.get(data[0]) is None:
              group_dict[data[0]] = {
                  "images": [{
                      "folder_name": data[1],
                      "image_id": data[2],
                      "image_name": data[3]
                  }]
              }
          else:
              group_dict[data[0]]["images"].append({
                  "folder_name": data[1],
                  "image_id": data[2],
                  "image_name": data[3]
              })
      return group_dict

delete_image: Takes image_list as input and deletes images from the database

  def delete_image(db: Session, image_list: dict):
      for key, value in image_list.items():
          for image in value:
              delete_image_from_database(db, key, image["image_id"])
      return 1

Now all the one drive connections and database connections can be established from the backend, I can now create an endpoint to access those operations or data. So I created a routing endpoint named image.

The functions for different operations and their actions are given below:

One Page Image List: This endpoint returns a maximum of 12 images. The function takes one input as page_no and then it communicates with the database through functions in the crud.py file and returns the low-resolution images

  @router.get("/page")
  async def get_whole_page_image(page: int = 1, db: Session = Depends(get_db)):
      if page < 1:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": -1,
                  "hint": "positive invalid page.. dont try this again"
              }
          )
      code, image_list = get_image_per_page(db, page)
      if code == 0:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 0,
                  "hint": "database_empty"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 1,
              "data": image_list,
              "hint": "got_images"
          }
      )

One Group Image List: This function endpoint accept group_no as input and returns all images contained in that group. The group images are fetched from the database again from the function inside the crud.py file.

  @router.get("/group")
  async def get_group_image(group: int = 1, db: Session = Depends(get_db)):
      image_list = get_image_from_group(db, group)
      if not image_list:
          return {
              "status": 0,
              "data": "invalid group.. dont try this again"
          }
      return {
          "status": 1,
          "data": image_list
      }

High-Resolution Image: This function endpoint accepts group_no and image_no as input and returns a single high-resolution image. The high-resolution image is fetched from the database again from the function inside the crud.py file.

  @router.get("/")
  async def get_single_image(group: int, image_id: int, db: Session = Depends(get_db)):
      image_info = get_one_image(db, group, image_id)
      if image_info is None:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 0,
                  "hint": "image_not_found"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 1,
              "data": image_info,
              "hint": 'image_found'
          }
      )

Total Page: This API endpoint is used to count all image groups and returns the possible page number. Each page can contain a maximum of 12 images.

  @router.get("/total-page")
  async def get_total_pages(db: Session = Depends(get_db)):
      total = get_last_group(db)
      pages = total // 12
      if total % 12 != 0:
          pages += 1
      return {
          "totalPage": pages
      }

Download Full Resolution Image: This function endpoint accepts group_no and image_no as input and returns the actual full-resolution image. The full-resolution image is fetched from the database again from the function inside the crud.py file.

  @router.get("/download")
  async def download_operation(group_id: int, image_id: int, db: Session = Depends(get_db)):
      image = download_full_image(db, group_id, image_id)
      if image is not None:
          return JSONResponse(
              status_code=status.HTTP_200_OK,
              content={
                  "code": 1,
                  "image": image,
                  "hint": "image_found"
              }
          )
      return JSONResponse(
          status_code=status.HTTP_200_OK,
          content={
              "code": 0,
              "hint": "image_not_found"
          }
      )

Database ImageDB Schema

Database ImageDB Data

Database Onedrive Schema

Database UserDB Schema

Database UserDB Data

The Fastapi server with all operations is added now. Now it can be fully connected to front-end and can be used successfully

Github URL Of The Project [Backend] :

https://github.com/DeepProgram/artetoreAPI

Building a Full Stack Web App: How I Developed A ImageHub With Python And React Part - 2 [Backend]

Subscribe to my newsletter

Deep Sarker

Deep Sarker