Magento 2 and Adobe Commerce Python FastAPI Graph QL microservices using PyGento

To create a Magento 2 GraphQL API using Python FastAPI to fetch Magento products with the CatalogProductEntity PyGento model, we'll use the graphene library for GraphQL schema definition and integration.

Install Required Dependencies

Make sure you have the required packages installed

pip install fastapi graphene uvicorn

What is Fast API for Magento ?

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python based on standard Python type hints.

The key features are:

• Fast: Very high performance, on par with NodeJS and Go (thanks to Starlette and Pydantic). One of the fastest Python frameworks available.
• Fast to code: Increase the speed to develop features by about 200% to 300%. *
• Fewer bugs: Reduce about 40% of human (developer) induced errors. *
• Intuitive: Great editor support. Completion everywhere. Less time debugging.
• Easy: Designed to be easy to use and learn. Less time reading docs.
• Short: Minimize code duplication. Multiple features from each parameter declaration. Fewer bugs.

Add PyGento from this GitHub to your project ->

GitHub - Genaker/PyGento: Python module to work with Magneto Database directly without using broken…

We will use catalog microservice and Product model.

from sqlalchemy import Column, DECIMAL, Date, DateTime, Float, ForeignKey, Index, String, TIMESTAMP, Table, Text, text
from sqlalchemy.dialects.mysql import BIGINT, INTEGER, MEDIUMTEXT, SMALLINT
from sqlalchemy.orm import relationship, backref
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
metadata = Base.metadata

class CatalogProductEntity(Base):
    __tablename__ = 'catalog_product_entity'
    __table_args__ = {'comment': 'Catalog Product Table'}

    entity_id = Column(INTEGER(10), primary_key=True, comment='Entity ID')
    attribute_set_id = Column(SMALLINT(5), nullable=False, index=True, server_default=text("0"), comment='Attribute Set ID')
    type_id = Column(String(32), nullable=False, server_default=text("'simple'"), comment='Type ID')
    sku = Column(String(64), index=True, comment='SKU')
    has_options = Column(SMALLINT(6), nullable=False, server_default=text("0"), comment='Has Options')
    required_options = Column(SMALLINT(5), nullable=False, server_default=text("0"), comment='Required Options')
    created_at = Column(TIMESTAMP, nullable=False, server_default=text("current_timestamp()"), comment='Creation Time')
    updated_at = Column(TIMESTAMP, nullable=False, server_default=text("current_timestamp() ON UPDATE current_timestamp()"), comment='Update Time')

    gallery = relationship('CatalogProductEntityMediaGallery', secondary='catalog_product_entity_media_gallery_value_to_entity')
    parents = relationship(
        'CatalogProductEntity',
        secondary='catalog_product_relation',
        primaryjoin='CatalogProductEntity.entity_id == catalog_product_relation.c.child_id',
        secondaryjoin='CatalogProductEntity.entity_id == catalog_product_relation.c.parent_id'
    )
    varchar = relationship("CatalogProductEntityVarchar", back_populates="entity")
    decimal = relationship("CatalogProductEntityDecimal", back_populates="entity")
    datetime = relationship("CatalogProductEntityDatetime", back_populates="entity")
    text = relationship("CatalogProductEntityText", back_populates="entity")
    intager = relationship("CatalogProductEntityInt", back_populates="entity")
    
    websites = relationship('StoreWebsite', secondary='catalog_product_website')

Models For EAV tables:

# Attribute values for 'varchar' type (e.g., 'name')
class CatalogProductEntityVarchar(Base):
    __tablename__ = 'catalog_product_entity_varchar'
    __table_args__ = (
        Index('CATALOG_PRODUCT_ENTITY_VARCHAR_ENTITY_ID_ATTRIBUTE_ID_STORE_ID', 'entity_id', 'attribute_id', 'store_id', unique=True),
        {'comment': 'Catalog Product Varchar Attribute Backend Table'}
    )

    value_id = Column(INTEGER(11), primary_key=True, comment='Value ID')
    attribute_id = Column(ForeignKey('eav_attribute.attribute_id', ondelete='CASCADE'), nullable=False, index=True, server_default=text("0"), comment='Attribute ID')
    store_id = Column(ForeignKey('store.store_id', ondelete='CASCADE'), nullable=False, index=True, server_default=text("0"), comment='Store ID')
    entity_id = Column(ForeignKey('catalog_product_entity.entity_id', ondelete='CASCADE'), nullable=False, server_default=text("0"), comment='Entity ID')
    value = Column(String(255), comment='Value')

    attribute = relationship('EavAttribute')
    entity = relationship('CatalogProductEntity', back_populates="varchar")
    store = relationship('Store')

# Attribute values for 'decimal' type (e.g., 'price')
class CatalogProductEntityDecimal(Base):
    __tablename__ = 'catalog_product_entity_decimal'
    value_id = Column(Integer, primary_key=True)
    entity_id = Column(Integer, ForeignKey('catalog_product_entity.entity_id'))
    attribute_id = Column(Integer)
    value = Column(Float)

# EAV attribute metadata
class EavAttribute(Base):
    __tablename__ = 'eav_attribute'
    attribute_id = Column(Integer, primary_key=True)
    attribute_code = Column(String)

Key Magento 2 Pyton components

GraphQL Schema:

• ProductType: Represents the structure of a product in the GraphQL API with fields like sku, name, price.
• Query: The root query defines the product field, which accepts an SKU as an argument to fetch product details.

FastAPI Integration:

• The GraphQLApp from starlette.graphql is used to add a GraphQL route (/graphql) to the FastAPI app.
• The schema specifies the query structure.

Resolver Function:

• The resolve_product method fetches the product using the CatalogProductEntity ORM model and maps its fields to the ProductType.

We are using Shared Database pattern: shared database across services while migrating a monolithic Magento Adobe Commerce to microservice.

A shared database acts as a transitional phase, allowing microservices to be developed incrementally while leveraging the existing monolithic magento frontend and admin and database.

FastAPI Application Code

Main App Code

from fastapi import FastAPI, Query
from starlette.graphql import GraphQLApp
from graphene import ObjectType, String, Float, Field, Schema
from models.catalog import CatalogProductEntity as Product

# Define GraphQL Object Type
class ProductType(ObjectType):
    sku = String()
    name = String()
    price = Float()
    status = String()
# Define Query Type
class Query(ObjectType):
    product = Field(ProductType, sku=String(required=True))
    def resolve_product(self, info, sku):
        # Fetch product by SKU
        product = Product.query.filter_by(sku=sku).first()
        // attribute_id 
        NAME = 33
        PRICE = 64
        status = 1
        if not product:
            return None

    # Add a filter directly on the relationship
    name = attr.value for attr in product.varchar.filter_by(attribute_id=NAME)
    ## example to loop all the attributes 
    price = next((attr.value for attr in product.decimal if attr.attribute_id == PRICE), None)

        # Map product fields
        return ProductType(
            sku=product.sku,
            name= name,
            price= price,
            status="Enabled" if status == 1 else "Disabled"
        )
# Create GraphQL Schema
schema = Schema(query=Query)
# Add GraphQL route to FastAPI
app.add_route("/graphql", GraphQLApp(schema=schema))

PyGento Relationships:

• varchar_attributes and decimal_attributes are assumed relationships in Pygento to fetch attribute values for varchar and decimal data types, respectively. These relationships allow you to dynamically access attributes without directly querying the EAV tables.
• Filter Attributes:
• Use Python’s next() to filter the specific attributes like name or price based on the attribute_code.

Running the Application

Run the FastAPI application with:

uvicorn app:app --reload

Open the GraphQL Playground at http://127.0.0.1:8000/graphql.

GraphQL Query Example

Use the following query in the GraphQL Playground to fetch a product:

query {
  product(sku: "sample-sku") {
    sku
    name
    price
    status
  }
}

Response Example

If the product exists, you’ll get a response like:

{
  "data": {
    "product": {
      "sku": "sample-sku",
      "name": "Sample Product",
      "price": 49.99,
      "status": "Enabled"
    }
  }
}

If the product doesn’t exist, the response will be:

{
  "data": {
    "product": null
  }
}

Simple NodeJent Performance Benchmark 350 concurrent users.

ab -k -c 350 -n 5000 http://<rpi ip>:8000/graphql

The above command sends 5k requests in total with a concurrency of 350 requests. One could tweak these numbers as per requirements.

FastAPI (Sync) — Python

FastAPI and Like ExpressJS utilize only one CPU core per instance. So, we have a result of 178 responses per second for a single core. By scaling a number of Fast API instances, we can multiply performance dramatically. we are running FastAPI's ASGI server (uvicorn) with default settings (1 worker). If we tweak the worker count to the number of available CPUs, there should be a significant jump in the performance. Magento 2 PHP server in similar circumstances can generate around 1 response per second ….

Adding API Key Authorization to the endpoint:

from fastapi import FastAPI, Depends, HTTPException, Header, status

app = FastAPI()

# get the API key
API_KEY = getMagentoAPIKey();

# Dependency to validate the API key
def verify_api_key(authorization: str = Header(...)):
    # Check if the Authorization header is properly formatted
    if not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authorization header",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # Extract the token and verify it
    token = authorization[len("Bearer "):]
    if token != API_KEY:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid API key",
        )
    return True

# A protected endpoint
@app.get("/protected")
async def protected_endpoint(authorized: bool = Depends(verify_api_key)):
    return {"message": "Access granted"}

Notes

• Error Handling: Enhance the resolver to return appropriate error messages or handle exceptions for invalid SKUs or database connectivity issues.
• Pagination and Filters: Extend the schema to support paginated product listings or additional filters like price range, category, etc.
• Deployment: For production, use a proper ASGI server like Gunicorn with Uvicorn workers and configure HTTPS.
• It is just an example of how to work with Python Pygento and Adobe Commerce and not a production code.

This setup gives you a fully functional GraphQL API for querying Magento products using FastAPI. also Let me know if you have any questions (egorshitikov@gmail.com)! 🚀