LinePay/main.py

import dotenv
import uuid
import os
import logging
from typing import AsyncContextManager, Optional
from fastapi import FastAPI, Request, responses
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
from dotenv import load_dotenv
from os.path import join, dirname
from linepay import LinePayApi
from starlette.responses import HTMLResponse

# dotenv
dotenv_path = join(dirname(__file__),'./env/.env')
load_dotenv(dotenv_path)

# logger (TBD)

# template
templates = Jinja2Templates(directory="templates")

# Line Pay Config
LINE_PAY_CHANNEL_ID = os.environ.get("LINE_PAY_CHANNEL_ID")
LINE_PAY_CHANNEL_SECRET = os.environ.get("LINE_PAY_CHANNEL_SECRET")
LINE_PAY_REQEST_BASE_URL = "https://{}".format(os.environ.get("HOST_NAME"))
line = LinePayApi(LINE_PAY_CHANNEL_ID, LINE_PAY_CHANNEL_SECRET, is_sandbox=True)

# CACHE
CACHE = {}


# Fastapi
app = FastAPI()



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

## Request
@app.get('/request', response_class=HTMLResponse)
async def pay_request(request: Request):
    order_id = str(uuid.uuid4())
    amount = 1200
    currency = "TWD"
    CACHE["order_id"] = order_id
    CACHE["amount"] = amount
    CACHE["currency"] = currency
    request_options ={
        "amount" : amount,
        "currency" : currency,
        "orderId" : order_id,
        "packages" : [
            {
                "id" : "早鳥方案",
                "amount" : 1200,
                "products" :[
                    {
                        # "id" : "Id_早鳥方案",
                        "name" : "早鳥方案",
                        "quantity" : 1,
                        "price" : 1200,
                        "imageUrl" : "https://kb.rspca.org.au/wp-content/uploads/2018/11/golder-retriever-puppy.jpeg"
                    }
                ]
            }

        ],
        "redirectUrls" : {
            "confirmUrl" : LINE_PAY_REQEST_BASE_URL + "/confirm/",
            "cancelUrl" : LINE_PAY_REQEST_BASE_URL + "/cancel/"
        }
    }
    response = line.request(request_options)
    transaction_id = int(response.get("info",{}).get("transactionId",0))
    check_result = line.check_payment_status(transaction_id)
    response["transaction_id"] = transaction_id
    response["paymentStatusCheckReturnCode"] = check_result.get("returnCode", None)
    response["paymentStatusCheckReturnMessage"] = check_result.get("returnMessage", None)
    return templates.TemplateResponse("request.html", {"request":response})
    # return response


## Confirm
@app.get('/confirm/')
async def pay_confirm(transactionId: int, orderId: Optional[str] = None):
    CACHE["transaction_id"] = transactionId
    response = line.confirm(transactionId,float(CACHE.get("amount",0)),CACHE.get("currency","TWD"))
    check_result = line.check_payment_status(transactionId)
    payment_details = line.payment_details(transaction_id=transactionId)
    response["transaction_id"] = transactionId
    response["paymentStatusCheckReturnCode"] = check_result.get("returnCode", None)
    response["paymentStatusCheckReturnMessage"] = check_result.get("returnMessage",None)
    response["payment_details"] = payment_details
    return templates.TemplateResponse("confirm.html", {"request":response})
    # return {"transactionId" : str(transactionId), "orderId" : orderId}
## Capture

## Refund

## Payment Details API
@app.get('/payments')
async def pay_payments(orderId : str):
    payment_details = line.payment_details(order_id=orderId)
    return payment_details

## Pay Preapproved API