LinePay/main.py

import dotenv
import uuid
import os
import logging
from typing import AsyncContextManager, Optional
from fastapi import FastAPI, Request, responses
from fastapi.params import Depends
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 pydantic.errors import UrlSchemeError
from starlette.responses import HTMLResponse
from pydantic import BaseModel, EmailStr, validator
from sqlalchemy.orm import Session
from . import crud, models, schemas
from .database import SessionLocal, engine




# dotenv
dotenv_path = join(dirname(__file__),'./env/.env')
# dotenv_path = join(dirname(__file__),'./env/test.env') ## sandbox
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=False)
# line = LinePayApi(LINE_PAY_CHANNEL_ID, LINE_PAY_CHANNEL_SECRET, is_sandbox=True)

# CACHE
CACHE = {}



# db
models.Base.metadata.create_all(bind=engine)

# Fastapi
app = FastAPI()


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

## Request
@app.post('/request', response_class=HTMLResponse)
async def pay_request(userinfo: schemas.UserInfo, db: Session = Depends(get_db)):
    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