from __future__ import annotations import datetime, hashlib, logging, os, pytz, redis, secrets from typing import Dict, List, Optional, Set, Union from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator from app.db_sql import redis_lookup_id_random, get_id_random from app.lib_general import log, logging from .common_field_schema import base_fields, default_num_bytes # Updated 2022-01-20 class Order_Line_Base(BaseModel): log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) id_random: Optional[str] = Field( **base_fields['order_line_id_random'], alias = 'order_line_id_random', default_factory = lambda:secrets.token_urlsafe(default_num_bytes), ) id: Optional[int] = Field( alias = 'order_line_id' ) order_id_random: Optional[str] order_id: Optional[int] product_id_random: str = None product_id: Optional[int] account_id_random: Optional[str] account_id: Optional[int] account_name: Optional[str] product_for_type: Optional[str] # Copied from product record product_for_id_random: Optional[str] # Copied from product record NOPE product_for_id: Optional[int] # Copied from product record product_type_id: Optional[int] # Copied from product record product_type: Optional[str] # WARNING: Copied from product record; dup from look up? probably not use? product_type_code: Optional[str] # Copied from product record; from look up product_type_name: Optional[str] # Copied from product record; from look up product_name: Optional[str] # Copied from product record product_description: Optional[str] # Copied from product record product_unit_price: Optional[int] # Copied from product record product_recurring: Optional[bool] # Copied from product record curr_product_id_random: Optional[str] # Should be the same as product_id_random above curr_product_id: Optional[int] # Should be the same as product_id above # NOTE: This is reversed with for_id_random curr_product_for_type: Optional[str] # Dynamic from v_order_line curr_product_for_id: Optional[int] # Dynamic from v_order_line curr_product_for_id_random: Optional[str] # Dynamic from v_order_line NOPE curr_product_type_id: Optional[int] # Dynamic from v_order_line curr_product_type: Optional[str] # Dynamic from v_order_line curr_product_type_code: Optional[str] # Dynamic from v_order_line curr_product_type_name: Optional[str] # Dynamic from v_order_line curr_product_name: Optional[str] # Dynamic from v_order_line curr_product_description: Optional[str] # Dynamic from v_order_line curr_product_unit_price: Optional[int] # Dynamic from v_order_line curr_product_max_quantity: Optional[int] # Dynamic from v_order_line curr_product_recurring: Optional[bool] # Dynamic from v_order_line for_person_id: Optional[int] for_person_id_random: Optional[str] for_person_given_name: Optional[str] # Dynamic from v_order_line for_person_family_name: Optional[str] # Dynamic from v_order_line for_person_display_name: Optional[str] # Dynamic from v_order_line for_person_full_name: Optional[str] # Dynamic from v_order_line name: Optional[str] # Should be the same as product_name above quantity: int = Field(0, ge=0, lt=150) amount: int = Field(0, ge=0, lt=1500000) total: int = Field(0, ge=0, lt=1500000) # Calculated with trigger recurring: Optional[bool] = False message: Optional[str] notes: Optional[str] created_on: Optional[datetime.datetime] = None updated_on: Optional[datetime.datetime] = None # Including convenience data # This is only for convenience. Probably going to keep unless it causes a problem. dollar_amount: Optional[str] # From SQL view dollar_total: int = Optional[str] # From SQL view order_status: Optional[str] order_notes: Optional[str] order_created_on: Optional[datetime.datetime] = None order_updated_on: Optional[datetime.datetime] = None remove_line: bool = False # Including other related objects # product: Optional[Union[Product_Base, None]] # Future use? # for_person: Optional[Union[Person_Base, None]] # Future use? _processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now) #@validator('order_line_id_random', always=True) def order_line_id_random_copy(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['id_random']: return values['id_random'] return None @validator('id', always=True) def order_line_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['id_random']: log.debug(values['id_random']) return redis_lookup_id_random(record_id_random=values['id_random'], table_name='order_line') return None @validator('order_id', always=True) def order_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['order_id_random']: return redis_lookup_id_random(record_id_random=values['order_id_random'], table_name='order') return None @validator('product_id', always=True) def product_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values.get('product_id_random', None): return redis_lookup_id_random(record_id_random=values['product_id_random'], table_name='product') return None @validator('product_for_id', always=True) def product_for_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['product_for_id_random'] and values['product_for_type']: return redis_lookup_id_random(record_id_random=values['product_for_id_random'], table_name=values['product_for_type']) return None # @validator('curr_product_for_id_random', always=True) # def curr_product_for_id_random_lookup(cls, v, values, **kwargs): # log.setLevel(logging.DEBUG) # log.debug(locals()) # if values['curr_product_for_id'] and values['curr_product_for_type']: # return redis_lookup_id(record_id=values['curr_product_for_id'], table_name=values['curr_product_for_type']) # return None @validator('curr_product_for_id_random', always=True) def curr_product_for_id_random_lookup(cls, v, values, **kwargs): log.setLevel(logging.DEBUG) log.debug(locals()) if values['curr_product_for_id'] and values['curr_product_for_type']: return get_id_random(record_id=values['curr_product_for_id'], table_name=values['curr_product_for_type']) return None # @validator('curr_product_for_id', always=True) # def curr_product_for_id_lookup(cls, v, values, **kwargs): # log.setLevel(logging.WARNING) # log.debug(locals()) # if values['curr_product_for_id_random'] and values['curr_product_for_type']: # return redis_lookup_id_random(record_id_random=values['curr_product_for_id_random'], table_name=values['curr_product_for_type']) # return None @validator('account_id', always=True) def account_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values.get('account_id_random', None): return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account') return None class Config: underscore_attrs_are_private = True allow_population_by_field_name = True fields = base_fields class Order_Line_DB_Base(BaseModel): log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) id_random: Optional[str] = Field( alias = 'order_line_id_random', ) id: Optional[int] = Field( alias = 'order_line_id' ) order_id_random: Optional[str] order_id: Optional[int] product_id_random: str product_id: Optional[int] product_for_type: Optional[str] # Copied from product record product_for_id_random: Optional[str] # Copied from product record NOPE product_for_id: Optional[int] # Copied from product record product_type_id: Optional[int] # Copied from product record product_name: Optional[str] # Copied from product record product_unit_price: Optional[int] # Copied from product record product_recurring: Optional[bool] # Copied from product record for_person_id: Optional[int] for_person_id_random: Optional[str] name: Optional[str] # Should be the same as product_name above quantity: int = Field(0, ge=0, lt=150) amount: int = Field(0, ge=0, lt=1500000) recurring: Optional[bool] = False recurring_period: Optional[int] message: Optional[str] notes: Optional[str] _processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now) class Config: underscore_attrs_are_private = True allow_population_by_field_name = True class Order_Line_Full_Detail_Base(Order_Line_Base): log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) person_id: Optional[int] person_id_random: Optional[str] person_given_name: Optional[str] person_family_name: Optional[str] person_display_name: Optional[str] person_full_name: Optional[str] person_contact_id: Optional[int] person_contact_id_random: Optional[str] person_contact_for_type: Optional[str] person_contact_for_id: Optional[int] person_contact_name: Optional[str] person_contact_email: Optional[str] person_contact_cc_email: Optional[str] person_contact_phone_mobile: Optional[str] person_contact_phone_home: Optional[str] person_contact_phone_office: Optional[str] person_contact_phone_land: Optional[str] person_contact_phone_fax: Optional[str] person_contact_phone_other: Optional[str] person_contact_address_id: Optional[int] person_contact_address_id_random: Optional[str] person_contact_address_for_type: Optional[str] person_contact_address_for_id: Optional[int] person_contact_address_name: Optional[str] person_contact_address_organization_name: Optional[str] person_contact_address_line_1: Optional[str] person_contact_address_line_2: Optional[str] person_contact_address_line_3: Optional[str] person_contact_address_city: Optional[str] person_contact_address_country_subdivision_code: Optional[str] person_contact_address_country_subdivision_name: Optional[str] # From country subdivision lookup table person_contact_address_state_province: Optional[str] # Avoid using person_contact_address_postal_code: Optional[str] person_contact_address_country_alpha_2_code: Optional[str] person_contact_address_country_name: Optional[str] # From country lookup table person_contact_address_country: Optional[str] # Avoid using person_contact_address_lu_time_zone_id: Optional[str] # Including convenience data # This is only for convenience. Probably going to keep unless it causes a problem. person_email: Optional[str] person_cc_email: Optional[str] # Maybe add timezone in the future? class Config: underscore_attrs_are_private = True allow_population_by_field_name = True Order_Line_Base.update_forward_refs()