Object of type Document is not JSON serializable

[jsvaldez829]

Trying this chainlit code for a conversationalQAretrieval and getting this error:

Exception ignored in: <async_generator object Runnable._atransform_stream_with_config at 0x0000024E082E2BC0>
Traceback (most recent call last):
  File "C:\Users\jsval\chainlit-cookbook\pinecone\myenv5\Lib\site-packages\langchain\schema\runnable\base.py", line 1581, in atransform
    yield chunk
RuntimeError: async generator ignored GeneratorExit
2023-11-18 08:59:52 - Object of type Document is not JSON serializable
Traceback (most recent call last):
  File "C:\Users\jsval\chainlit-cookbook\pinecone\myenv5\Lib\site-packages\chainlit\utils.py", line 39, in wrapper
    return await user_function(**params_values)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\jsval\chainlit-cookbook\pinecone\app6.py", line 110, in on_message
    await msg.stream_token(chunk)
  File "C:\Users\jsval\chainlit-cookbook\pinecone\myenv5\Lib\site-packages\chainlit\message.py", line 158, in stream_token
    await context.emitter.send_token(
  File "C:\Users\jsval\chainlit-cookbook\pinecone\myenv5\Lib\site-packages\socketio\async_server.py", line 168, in emit
    await self.manager.emit(event, data, namespace, room=room,
  File "C:\Users\jsval\chainlit-cookbook\pinecone\myenv5\Lib\site-packages\socketio\async_manager.py", line 38, in emit
    encoded_packet = pkt.encode()
                     ^^^^^^^^^^^^
  File "C:\Users\jsval\chainlit-cookbook\pinecone\myenv5\Lib\site-packages\socketio\packet.py", line 64, in encode
    encoded_packet += self.json.dumps(data, separators=(',', ':'))
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.11_3.11.1776.0_x64__qbz5n2kfra8p0\Lib\json\__init__.py", line 238, in dumps
    **kw).encode(obj)
          ^^^^^^^^^^^
  File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.11_3.11.1776.0_x64__qbz5n2kfra8p0\Lib\json\encoder.py", line 200, in encode
    chunks = self.iterencode(o, _one_shot=True)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.11_3.11.1776.0_x64__qbz5n2kfra8p0\Lib\json\encoder.py", line 258, in iterencode
    return _iterencode(o, 0)
           ^^^^^^^^^^^^^^^^^
  File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.11_3.11.1776.0_x64__qbz5n2kfra8p0\Lib\json\encoder.py", line 180, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type Document is not JSON serializable
2023-11-18 08:59:53 - HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"

This is the code:

import os
import pinecone
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser, format_document
from langchain.schema.runnable import Runnable, RunnableMap, RunnablePassthrough, RunnableLambda
from langchain.vectorstores import Pinecone
from langchain.memory import ConversationBufferMemory
from typing import List, Tuple
from langchain.embeddings.openai import OpenAIEmbeddings
from operator import itemgetter
from langchain.schema.runnable.config import RunnableConfig
import getpass

from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter

import chainlit as cl

# Initialize Pinecone
pinecone.init(
    api_key=os.environ.get("PINECONE_API_KEY"),
    environment=os.environ.get("PINECONE_ENV"),
)

embeddings = OpenAIEmbeddings()

namespace = None

# Pinecone index name
index_name = "chatleyes"

# Create Pinecone retriever
vectorstore = Pinecone.from_existing_index(
        index_name=index_name, embedding=embeddings, namespace=namespace
    )
retriever = vectorstore.as_retriever(search_kwargs={'k': 7})

# Functions to format chat history and combine documents
def _format_chat_history(chat_history: List[Tuple]) -> str:
    buffer = ""
    for dialogue_turn in chat_history:
        human = "Human: " + dialogue_turn[0]
        ai = "Assistant: " + dialogue_turn[1]
        buffer += "\n" + "\n".join([human, ai])
    return buffer

def _combine_documents(docs, document_prompt=ChatPromptTemplate.from_template(template="{page_content}"), document_separator="\n\n"):
    doc_strings = [format_document(doc, document_prompt) for doc in docs]
    return document_separator.join(doc_strings)

# Memory setup
memory = ConversationBufferMemory(return_messages=True, output_key="answer", input_key="question")

# Chat start and message handling with RAG and memory
@cl.on_chat_start
async def on_chat_start():
    loaded_memory = RunnablePassthrough.assign(
        chat_history=RunnableLambda(memory.load_memory_variables) | itemgetter("history"),
    )

    standalone_question = {
        "standalone_question": {
            "question": lambda x: x["question"],
            "chat_history": lambda x: _format_chat_history(x["chat_history"]),
        }
        | ChatPromptTemplate.from_template("Rephrase question: {question}") # Modify as needed
        | ChatOpenAI(temperature=0)
        | StrOutputParser(),
    }

    retrieved_documents = {
        "docs": itemgetter("standalone_question") | retriever,
        "question": lambda x: x["standalone_question"],
    }

    final_inputs = {
        "context": lambda x: _combine_documents(x["docs"]),
        "question": itemgetter("question"),
    }

    answer = {
        "answer": final_inputs | ChatPromptTemplate.from_template("Answer: {context}\nQuestion: {question}") | ChatOpenAI(), # Modify as needed
        "docs": itemgetter("docs"),
    }

    final_chain = loaded_memory | standalone_question | retrieved_documents | answer
    cl.user_session.set("runnable", final_chain)
    
    
@cl.on_message
async def on_message(message: cl.Message):
    runnable = cl.user_session.get("runnable")  # type: Runnable

    msg = cl.Message(content="")

    async for chunk in runnable.astream(
        {"question": message.content},
        config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()]),
    ):
        await msg.stream_token(chunk)

    await msg.send()

[dosubot[bot]]

🤖

Hello @jsvaldez829! I'm Dosu, a bot designed to assist you with troubleshooting, answering questions, and guiding you towards becoming a contributor. While we wait for a human maintainer, feel free to let me know how I can help you further.

The error you're encountering, "TypeError: Object of type Document is not JSON serializable", suggests that you're trying to serialize a Document object that isn't inherently serializable. This is likely happening when you're trying to send the Document object over a network or save it in a format that requires serialization, such as JSON.

In the context of LangChain, a Document object is a class for storing a piece of text and associated metadata. It inherits from the Serializable class. Here is the structure of the Document object:

class Document(Serializable):
    """Class for storing a piece of text and associated metadata."""

    page_content: str
    """String text."""
    metadata: dict = Field(default_factory=dict)
    """Arbitrary metadata about the page content (e.g., source, relationships to other
        documents, etc.).
    """
    type: Literal["Document"] = "Document"

    @classmethod
    def is_lc_serializable(cls) -> bool:
        """Return whether this class is serializable."""
        return True

(Source: libs/langchain/langchain/schema/document.py)

From the traceback you provided, it seems like the error is occurring when you're trying to send a Document object over a SocketIO connection. SocketIO uses JSON to serialize data, and since the Document object isn't JSON serializable, it's causing an error.

To fix this issue, you need to convert the Document object into a format that is JSON serializable before sending it over the SocketIO connection. You can do this by accessing the page_content and metadata properties of the Document object and sending those instead. Here's an example of how you can do this:

# Assuming `doc` is a Document object
json_serializable_doc = {
    'page_content': doc.page_content,
    'metadata': doc.metadata
}

You can then send json_serializable_doc over the SocketIO connection.

I hope this helps! Let me know if you have any other questions.

Sources

• TypeError: Object of type Document is not JSON serializable
• libs/langchain/langchain/load/dump.py
• libs/langchain/langchain/retrievers/google_cloud_documentai_warehouse.py
• libs/langchain/langchain/schema/document.py

---

This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

[tituslhy]

Sorry so how do we amend the code snippet above, please? I believe @jsvaldez829 was following an official langchain cookbook recipe for retrieval augmented generation?

[mrctito]

Hello,

I have the same code and I'm facing the same issue. However, I disagree with the solution proposed by DOSU because if I do it like this, it works:

inputs = {"question": message.content}
result = await runnable.ainvoke(inputs)
msg = cl.Message(content=result["answer"], disable_feedback=True)
await msg.send()

In other words, the response is in result["answer"].

The error only occurs when I try to do the streaming.

Thank you.

[ryann-sportsbet]

Did you figure it out how to work with astream? @mrctito

[eveliotc]

I ran into this today and found a hacky workaround for this:

The reason this happens are actually three:

1. There is no clean way to provide JSONEncoders for any type if json.dumps is used by a dependency directly, in my case socketio-python.
2. json.dumps has some interesting logic around the encoder to pick when certain parameters are given, in my case socketio-python passes separators=(',', ':') seemingly to make the payload smaller, so json attempts to use _default_encoder.
3. I suppose you could use a dictionary or such to make json happy instead of Document but then everywhere Document is used will bark at you saying it is not actually a Document object and can't access attributes e.g. page_content

Anyhow the hacky workaround is:

1. Carry on using Document as as usual, this applies to other types you run into.
2. Define your own JSONEncoder to support the type you need, like this

class DocumentEncoder(json.JSONEncoder):

  def __init__(self):
    self.__dict__.update(json._default_encoder.__dict__)
    self._default_encoder = json._default_encoder

  def default(self, obj):
    if isinstance(obj, Document):
      return {   # `obj.__dict__`, `Serializable.to_json` or similar could also work here
        "page_content": obj.page_content,
        "metadata": obj.metadata
      }
    return self._default_encoder.default(self, obj)

3. here is where the hacky bits start: monkey patch json._default_encoder, like this:

# Workaround chainlit/socketio serialization
json._default_encoder = DocumentEncoder()

Keep in mind you should only do this once so you can use the same JSONEncoder implementation for all your types. Also make sure to pray no other dependency or piece of code hacks their way like this too...

4. Also patch your dependencies to force them to use the now provided default encoder, for instance for socketio-python:

# patch socketio packet json encoding so that json._default_encoder is used
sed -i "s/, separators=(',', ':'))/)/g" .venv/lib/python*/site-packages/socketio/packet.py

This patching woulb be needed every time the library is updated unless you use a fork
5. Have fun and good luck.

This is brittle so for a real solution libraries (including langchain) should be properly updated to allow users to provide JSONEncoders for their types somehow or even bring your own json encoding method/classes. Perhaps the python standard library too but not holding my breath on that one :)

[KingPungy]

As above chain.ainvoke({}) works but wont stream.

My solution is as follows:

cb = cl.LangchainCallbackHandler()
async for chunk in chain.astream({"input": question, "chat_history": message_history.messages},
                                 config=RunnableConfig(callbacks=[cb])):
        if "answer" in chunk:  # check if the chunk has an answer key
            await msg.stream_token(chunk["answer"])
await msg.send()

Hope this helped.

I use ChatOllama in my project and its not the fastest. (it never was)

NOTE if you are looking for performance this solutions might not be the best.

[eyurtsev]

1. Try to determine what is adding non serializable content to the document. There's likely non serializable content in the metadata. Potentially a retriever or a vectorstore is inserting something it shouldn't into the metadata.
2. You can use astream_events v2 for streaming if you need to stream from intermediate steps. You have full control over the data you get, so you can clean it up and serialize it yourself prior to sending it over the wire.

[Kenedy4]

I have same error:
import os
from functools import wraps
from flask import Flask, config, jsonify, request, session, send_from_directory
from flask_restful import Api, Resource
from models import db, Student, Teacher, Parent, Class, Subject, Notifications, LearningMaterial, PasswordResetToken
from config import Config
from flask_mail import Mail
from werkzeug.security import generate_password_hash
from werkzeug.utils import secure_filename
from flask_migrate import Migrate # type: ignore
from flask_bcrypt import Bcrypt # type: ignore
from flask_jwt_extended import JWTManager # type: ignore
import datetime
import secrets
from flask_mail import Message

from auth import CheckSession, Login, Logout, Signup

from flask_cors import CORS

app = Flask(name)
app.config.from_object(Config)

Initializing necessary extensions extensions

migrate = Migrate(app, db)
bcrypt = Bcrypt(app)
jwt = JWTManager(app)
mail = Mail(app)
CORS = (app)
api = Api(app)

Initializing the database

db.init_app(app)

File upload configurations

UPLOAD_FOLDER = 'uploads'
ALLOWED_EXTENSIONS = {'pdf', 'docx', 'txt'}

Adds file upload configuration inside the app context

with app.app_context():
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER

# Ensures the 'uploads' folder exists
if not os.path.exists(app.config['UPLOAD_FOLDER']):
    os.makedirs(app.config['UPLOAD_FOLDER'])

Helper function to check allowed file extensions

def allowed_file(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

Helper function to check if a user is logged in

def login_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if 'parent_id' not in session:
return jsonify({"error": "Unauthorized access, please log in"}), 401
return f(*args, **kwargs)
return decorated_function

Token required decorator

def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = request.headers.get('Authorization')
if not token or not token.startswith("Bearer "):
return jsonify({'message': 'Token is missing or in incorrect format!'}), 403

    try:
        token = token.split()[1]  # Extract token part
        decoded_token = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
        current_user = Teacher.query.get(decoded_token['user_id'])
        if not current_user:
            return jsonify({'message': 'User not found!'}), 404
    except Exception as e:
        return jsonify({'message': f'Error processing token: {str(e)}'}), 400

    return f(current_user, *args, **kwargs)

return decorated

Error handler for internal server errors

@app.errorhandler(500)
def internal_server_error(e):
return jsonify({"error": "Internal server error", "message": str(e)}), 500

Routes Using RESTful API

            ## Home Route #####

class Welcome(Resource):
def get(self):
return jsonify({"message": "Welcome to SecLink"}), 200

Auth Routes

class Signup(Resource):
def post(self):
data = request.get_json()
username = data.get('username')
password = data.get('password')
role = data.get('role')

    if role == 'teacher':
        new_user = Teacher(username=username, email=data.get('email'))
    elif role == 'parent':
        new_user = Parent(username=username, email=data.get('email'))
    else:
        return jsonify({'message': 'Invalid role.'}), 400

    hashed_password = bcrypt.generate_password_hash(password).decode('utf-8')
    new_user.password = hashed_password

    db.session.add(new_user)
    db.session.commit()

    return jsonify({'message': f'{role.capitalize()} registered successfully.'}), 201

class Login(Resource):
def post(self):
data = request.get_json()
username = data.get('username')
password = data.get('password')

    user = User.query.filter_by(username=username).first()
    if user and bcrypt.check_password_hash(user.password, password):
        session['user_id'] = user.id
        session['role'] = 'teacher' if isinstance(user, Teacher) else 'parent'  # Set role

        token = create_access_token(identity=user.id, expires_delta=datetime.timedelta(hours=1))

        return jsonify({
            'id': user.id,
            'username': user.username,
            'role': session['role'],
            'token': token
        }), 200
    
    return jsonify({'error': 'Invalid credentials'}), 401

class CheckSession(Resource):
@login_required
def get(self):
user_id = session.get('user_id')
if not user_id:
return jsonify({'error': 'No active session'}), 401

    # Retrieve the user by ID
    user = User.query.get(user_id)
    if not user:
        return jsonify({'error': 'User not found'}), 404
    
    role = 'teacher' if isinstance(user, Teacher) else 'parent'
    
    return {
        'id': user.id,
        'username': user.username,
        'email': user.email,
        'role': role
    }, 200

class Logout(Resource):
def delete(self):
session.pop('user_id', None)
session.pop('role', None)
return {}, 204
## User Route ######

class User(Resource):
@token_required
def get(self, user_id=None):
if user_id:
# Fetch specific user by ID
user = User.query.get(user_id)
if not user:
return jsonify({'error': 'User not found'}), 404
return jsonify(user.to_dict()), 200
else:
# Fetch all users
users = User.query.all()
return jsonify([user.to_dict() for user in users]), 200

@token_required
def post(self):
    data = request.get_json()
    role = data.get('role')
    username = data.get('username')
    password = data.get('password')
    
    if role == 'teacher':
        new_user = Teacher(username=username)
    elif role == 'parent':
        new_user = Parent(username=username)
    else:
        return jsonify({'message': 'Invalid role.'}), 400

    # Hash the password before storing
    new_user.password = data.get('password')
    
    # Add the new user to the database
    db.session.add(new_user)
    db.session.commit()
    
    return jsonify({'message': f'{role.capitalize()} registered successfully.'}), 201

@token_required
def get(self, role):
    if role == 'parent':
        users = Parent.query.all()
    elif role == 'teacher':
        users = Teacher.query.all()
    else:
        return jsonify({'error': 'Invalid role provided'}), 400

    return jsonify([user.to_dict() for user in users]), 200

            ## Parent Route ##

class Parent(Resource):
@token_required
def get(self, parent_id=None):
if parent_id:
parent = Parent.query.get_or_404(parent_id)
return jsonify(parent.to_dict()), 200
else:
parents = Parent.query.all()
return jsonify([parent.to_dict() for parent in parents]), 200

class Notifications(Resource):

@token_required

def get(self):

parent_id = session.get('parent_id')

parent = Parent.query.get_or_404(parent_id)

notifications = Notifications.query.filter_by(parent_id=parent.id).all()

return jsonify([notification.to_dict() for notification in notifications]), 200

class Student(Resource):

@token_required

def get(self):

parent_id = session.get('parent_id')

parent = Parent.query.get_or_404(parent_id)

students = Student.query.filter_by(parent_id=parent.id).all()

return jsonify([student.to_dict() for student in students]), 200

            ## Teacher Route ##

class Teacher(Resource):
@login_required
def get(self, teacher_id=None):
if teacher_id:
teacher = Teacher.query.get_or_404(teacher_id)
return jsonify(teacher.to_dict()), 200
teachers = Teacher.query.all()
return jsonify([teacher.to_dict() for teacher in teachers]), 200

@login_required
@token_required
def post(self):
    data = request.get_json()
    teacher = Teacher(
        username=data.get('username'),
        email=data.get('email'),
    )
    teacher.password = generate_password_hash(data.get('password'))

    db.session.add(teacher)
    db.session.commit()
    return jsonify(teacher.to_dict()), 201

@login_required
@token_required
def put(self, teacher_id):
    data = request.get_json()
    teacher = Teacher.query.get_or_404(teacher_id)
    teacher.username = data.get('username')
    teacher.email = data.get('email')
    db.session.commit()
    return jsonify(teacher.to_dict()), 200

@login_required
@token_required
def delete(self, teacher_id):
    teacher = Teacher.query.get_or_404(teacher_id)
    db.session.delete(teacher)
    db.session.commit()
    return '', 204
            ##  Student Route ##

class Student (Resource):
@token_required
def get(self, student_id=None):
if student_id:
student = Student.query.get_or_404(student_id)
return jsonify(student.to_dict()), 200
else:
students = Student.query.all()
return jsonify([student.to_dict() for student in students]), 200

            ##  LearningMaterial Route ##

class LearningMaterial(Resource):
@login_required
@token_required
def get(self, learning_material_id=None):
if learning_material_id:
learning_material = LearningMaterial.query.get_or_404(learning_material_id)
return learning_material.to_dict(), 200
learning_materials = LearningMaterial.query.all()
return jsonify([learning_material.to_dict() for learning_material in learning_materials]), 200

@login_required
def delete(self, learning_material_id):
    learning_material = LearningMaterial.query.get_or_404(learning_material_id)
    db.session.delete(learning_material)
    db.session.commit()
    return '', 204
            ##  LearningMaterialUpload Route ##

class LearningMaterialUpload(Resource):
@login_required
@token_required
def post(self):
teacher_id = session.get('user_id')
teacher = Teacher.query.get(teacher_id)

    if 'file' not in request.files:
        return jsonify({"error": "No file part"}), 400

    file = request.files['file']
    if file.filename == '':
        return {"error": "No selected file"}, 400

    if file and allowed_file(file.filename):
        filename = secure_filename(file.filename)
        file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))

        data = request.form
        learning_material = LearningMaterial(
            title=data.get('title'),
            file_path=filename,
            teacher_id=teacher.id,
            subject_id=data.get('subject_id')
        )

        db.session.add(learning_material)
        db.session.commit()

        return {"message": "File uploaded successfully"}, 201

    return {"error": "Invalid file format"}, 400

class LearningMaterialDownload(Resource):
@login_required
@token_required
def get(self, learning_material_id):
learning_material = LearningMaterial.query.get_or_404(learning_material_id)
file_path = learning_material.file_path
file_name = os.path.basename(file_path)

    try:
        return send_from_directory(app.config['UPLOAD_FOLDER'], file_name, as_attachment=True)
    except FileNotFoundError:
        return jsonify({"error": "File not found"}), 404
    
            ##  Dashboard Route ##

class Dashboard(Resource):

@token_required

def get(self):

parent_id = session.get('parent_id')

parent = Parent.query.get_or_404(parent_id)

notifications = Notifications.query.filter_by(parent_id=parent.id).all()

students = Student.query.filter_by(parent_id=parent.id).all()

learning_materials = LearningMaterial.query.filter_by(parent_id=parent.id).all()

return {

'notifications': [notification.to_dict() for notification in notifications],

'students': [student.to_dict() for student in students],

'learning_materials': [learning_material.to_dict() for learning_material in learning_materials]

}, 200

            ##  Notification Route ##

class Notifications(Resource):
@login_required
@token_required
def get(self, notification_id=None):
if notification_id:
notification = Notifications.query.get_or_404(notification_id)
return notification.to_dict(), 200
notifications = Notifications.query.all()
return jsonify([notification.to_dict() for notification in notifications]), 200

@login_required
def post(self):
    data = request.get_json()
    notification = Notifications(
        message=data.get('message'),
    )
    db.session.add(notification)
    db.session.commit()
    return notification.to_dict(), 201

@login_required
def delete(self, notification_id):
    notification = Notifications.query.get_or_404(notification_id)
    db.session.delete(notification)
    db.session.commit()
    return '', 204

            ##  Class Route ##

class Class(Resource):
@token_required
@login_required
def get(self, class_id=None):
if class_id:
# Fetch the class by class_id
class_obj = Class.query.get(class_id)
if not class_obj:
return jsonify({'message': 'Class not found.'}), 404
return jsonify(class_obj.to_dict()), 200
else:
# Fetch all classes if no class_id is provided
classes = Class.query.all()
return jsonify([class_obj.to_dict() for class_obj in classes]), 200
## Subject Route ##
class Subject(Resource):
@login_required
@token_required
def get(self, subject_id=None):
if subject_id:
subject = Subject.query.get_or_404(subject_id)
return subject.to_dict(), 200
subjects = Subject.query.all()
return jsonify([subject.to_dict() for subject in subjects]), 200

@login_required
def post(self):
    data = request.get_json()
    subject = Subject(
        subject_name=data.get('subject_name'),
        subject_code=data.get('subject_code'),
        class_id=data.get('class_id'),
        teacher_id=data.get('teacher_id'),
    )
    db.session.add(subject)
    db.session.commit()
    return subject.to_dict(), 201

@login_required
def put(self, subject_id):
    data = request.get_json()
    subject = Subject.query.get_or_404(subject_id)
    subject.subject_name = data.get('subject_name')
    subject.subject_code = data.get('subject_code')
    db.session.commit()
    return subject.to_dict(), 200

@login_required
def delete(self, subject_id):
    subject = Subject.query.get_or_404(subject_id)
    db.session.delete(subject)
    db.session.commit()
    return '', 204

        ## PasswordResetting Route ##

class RequestPasswordReset(Resource):
def post(self):
data = request.get_json()
email = data.get('email')

    # Check if the user exists
    user = User.query.filter_by(email=email).first()
    if not user:
        return jsonify({'message': 'User not found'}), 404

    # Generate a unique reset token
    token = secrets.token_urlsafe(20)  # A secure random token
    expiry = datetime.datetime.utcnow() + datetime.timedelta(hours=1)  # Token valid for 1 hour

    # Save the token in the database
    reset_token = PasswordResetToken(user_id=user.id, token=token, expiry_date=expiry)
    db.session.add(reset_token)
    db.session.commit()

    # Send reset token via email
    reset_link = f"{app.config['FRONTEND_URL']}/reset-password?token={token}"
    msg = Message('Password Reset Request', recipients=[email])
    msg.body = f"Use the following link to reset your password: {reset_link}"
    email.send(msg)

    return jsonify({'message': 'Password reset link sent to your email.'}), 200

class PasswordResetConfirm(Resource):
def post(self):
data = request.get_json()
token = data.get('token')
new_password = data.get('new_password')

    # Find the token in the database
    reset_token = PasswordResetToken.query.filter_by(token=token).first()

    if not reset_token:
        return jsonify({'message': 'Invalid or expired token'}), 400

    # Check if the token is expired
    if reset_token.expiry_date < datetime.datetime.utcnow():
        return jsonify({'message': 'Token has expired'}), 400

    # Update the user's password
    user = User.query.get(reset_token.user_id)
    user.password = generate_password_hash(new_password)
    db.session.commit()

    # Delete the token after use
    db.session.delete(reset_token)
    db.session.commit()

    return jsonify({'message': 'Password has been updated successfully.'}), 200

Adding resources (endpoints)

api.add_resource(Welcome, '/', '/welcome')
api.add_resource(Signup, '/signup')
api.add_resource(Login, '/login')
api.add_resource(CheckSession, '/check-session')
api.add_resource(Logout, '/logout')

# Password reset endpoints

api.add_resource(RequestPasswordReset, '/password-reset-request')
api.add_resource(PasswordResetConfirm, '/password-reset-confirm')

# Other resources

api.add_resource(User, '/users', '/users/int:user_id')
api.add_resource(Parent, '/parents', '/parents/int:parent_id')
api.add_resource(Student, '/students', '/students/int:student_id')
api.add_resource(Teacher, '/teachers', '/teachers/int:teacher_id')
api.add_resource(Class, '/classes', '/classes/int:class_id')
api.add_resource(Subject, '/subjects', '/subjects/int:subject_id')
api.add_resource(Notifications, '/notifications', '/notifications/int:notification_id')
api.add_resource(LearningMaterial, '/learning_materials', '/learning_materials/int:learning_material_id')
api.add_resource(LearningMaterialUpload, '/upload')
api.add_resource(LearningMaterialDownload, '/download/int:learning_material_id')

api.add_resource(Dashboard, '/dashboard')

app execution point

if name == 'main':
app.run(port=Config.PORT) #debug=True)