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