admin/partsERP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.started removal of CDNs.\n 2. Improved modular structure for all parts of project including database.

Browse Source
This commit is contained in:
teddy
2024-08-30 23:27:28 +01:00
parent 18977c8cac
commit f9cd9ec33a
2895 changed files with 490579 additions and 7561 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
datastores/datastore_store_stripe.py Normal file
View File

@@ -0,0 +1,202 @@
"""
Project: PARTS Website
Author: Edward Middleton-Smith
Precision And Research Technology Systems Limited
Technology: DataStores
Feature: Store Stripe DataStore
Description:
Datastore for Store Stripe service
"""
# internal
# from routes import bp_home
import lib.argument_validation as av
from business_objects.store.basket import Basket, Basket_Item
from business_objects.store.product_category import Container_Product_Category, Product_Category
from business_objects.store.currency import Currency
from business_objects.store.image import Image
from business_objects.store.delivery_option import Delivery_Option
from business_objects.store.delivery_region import Delivery_Region
from business_objects.store.discount import Discount
from business_objects.store.order import Order
from business_objects.store.product import Product, Product_Permutation, Product_Price, Product_Filters
from business_objects.sql_error import SQL_Error
from business_objects.store.stock_item import Stock_Item, Stock_Item_Filters
from business_objects.user import User, User_Filters, User_Permission_Evaluation
from business_objects.store.product_variation import Product_Variation, Product_Variation_Filters, Product_Variation_List
from datastores.datastore_store_base import DataStore_Store_Base
from helpers.helper_db_mysql import Helper_DB_MySQL
# from models.model_view_store_checkout import Model_View_Store_Checkout # circular!
from extensions import db
# external
# from abc import ABC, abstractmethod, abstractproperty
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text
import stripe
import os
from flask import Flask, session, current_app
from pydantic import BaseModel, ConfigDict
from typing import ClassVar
from datetime import datetime
class DataStore_Store_Stripe(DataStore_Store_Base):
# Global constants
# Attributes
key_public_stripe: str = None
key_secret_stripe: str = None
def __init__(self):
super().__init__()
self.key_secret_stripe = os.environ.get("KEY_SECRET_STRIPE")
self.key_public_stripe = os.environ.get("KEY_PUBLIC_STRIPE")
# For sample support and debugging, not required for production:
stripe.set_app_info(
'stripe-samples/checkout-one-time-payments',
version='0.0.1',
url='https://github.com/stripe-samples/checkout-one-time-payments')
stripe.api_key = self.key_secret_stripe
def get_many_stripe_product_new(self):
_m = 'DataStore_Store_Stripe.get_many_stripe_product_new'
_m_db = 'p_shop_get_many_stripe_product_new'
# av.val_str(id_user)
# validation conducted by server
argument_dict_list = {
'a_id_user': self.info_user
}
print(f'executing {_m_db}')
result = self.db_procedure_execute(_m_db, argument_dict_list)
cursor = result.cursor
print('data received')
# Products
cursor.nextset()
result_set_1 = cursor.fetchall()
products = []
for row in result_set_1:
new_product = Product.from_DB_Stripe_product(row) # Product(row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7], row[8], row[9], row[10], row[11], row[12], row[13], row[14], row[15], row[16], row[17], row[18], row[19])
products.append(new_product)
print(f'products: {products}')
# Errors
cursor.nextset()
result_set_e = cursor.fetchall()
print(f'raw errors: {result_set_e}')
if len(result_set_e) > 0:
errors = [SQL_Error.from_DB_record(row) for row in result_set_e] # [SQL_Error(row[0], row[1]) for row in result_set_e]
for error in errors:
print(f"Error [{error.code}]: {error.msg}")
DataStore_Store_Stripe.db_cursor_clear(cursor)
return products
def get_many_stripe_price_new(self):
_m = 'DataStore_Store_Stripe.get_many_stripe_price_new'
_m_db = 'p_shop_get_many_stripe_price_new'
# av.val_str(id_user)
# validation conducted by server
argument_dict_list = {
'a_id_user': self.info_user
}
print(f'executing {_m_db}')
result = self.db_procedure_execute(_m_db, argument_dict_list)
cursor = result.cursor
print('data received')
# Products
cursor.nextset()
result_set_1 = cursor.fetchall()
products = []
for row in result_set_1:
new_product = Product.from_DB_Stripe_price(row) # Product(row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7], row[8], row[9], row[10], row[11], row[12], row[13], row[14], row[15], row[16], row[17], row[18], row[19])
products.append(new_product)
print(f'products: {products}')
# Errors
cursor.nextset()
result_set_e = cursor.fetchall()
print(f'raw errors: {result_set_e}')
if len(result_set_e) > 0:
errors = [SQL_Error.from_DB_record(row) for row in result_set_e] # [SQL_Error(row[0], row[1]) for row in result_set_e]
for error in errors:
print(f"Error [{error.code}]: {error.msg}")
DataStore_Store_Stripe.db_cursor_clear(cursor)
return products
def get_many_product_new(self):
_m = 'DataStore_Store_Stripe.get_many_product_new'
# Stripe
new_products = self.get_many_stripe_product_new()
for product in new_products:
product.id_stripe_product = self.create_stripe_product(product)
return new_products
def get_many_price_new(self):
_m = 'DataStore_Store_Stripe.get_many_product_new'
# Stripe
new_products = self.get_many_stripe_price_new()
for product in new_products:
product.id_stripe_price = self.create_stripe_price(product)
return new_products
# Stripe
def create_stripe_product(self, product): # _name, product_description):
_m = 'DataStore_Store_Stripe_Checkout.create_stripe_product'
# av.val_str(product_name, 'product_name', _m)
# av.val_str(product_description, 'product_description', _m)
av.val_instance(product, 'product', _m, Product)
print(f'stripe.api_key = {stripe.api_key}')
new_product = stripe.Product.create(
name = product.name,
description = product.description,
)
# Save these identifiers
print(f"Success! Here is your new Stripe product id: {new_product.id}")
return new_product.id
def create_stripe_price(self, product, currency): # product_id, product_price, product_currency, product_is_subscription, product_recurring_interval = '', product_interval_count = 0):
_m = 'DataStore_Store_Stripe_Checkout.create_stripe_price'
"""
av.val_str(p_id, 'p_id', _m)
av.full_val_float(p_price, 'p_price', _m, 0.01)
p_price = round(p_price, 2)
av.val_str(p_currency, 'p_currency', _m)
av.full_val_bool(p_is_subscription, 'p_is_subscription', _m)
p_is_subscription = bool(p_is_subscription)
av.val_str(p_recurring_interval, 'p_recurring_interval', _m)
av.full_val_int(p_interval_count, 'p_interval_count', _m, 1 if p_is_subscription else 0)
p_interval_count = int(p_interval_count)
"""
av.val_instance(product, 'product', _m, Product)
av.val_str(currency, 'currency', _m)
print(f'stripe.api_key = {stripe.api_key}')
new_product_price = stripe.Price.create(
unit_amount = product.unit_price,
currency = currency,
recurring = { "interval": product.name_recurring_interval, "interval_count": product.count_recurring_interval } if product.is_subscription else None,
product = product.id_stripe_product
)
# Save these identifiers
print(f"Success! Here is your Stripe product price id: {new_product_price.id} for {product.name}")
return new_product_price.id