75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from pyjeeves.models.raw import Company as CompanyModel, Customer as CustomerModel
|
|
from pyjeeves.models import db
|
|
from sqlalchemy.sql.expression import and_
|
|
from sqlalchemy.orm.strategy_options import Load
|
|
from sqlalchemy.orm.exc import NoResultFound
|
|
|
|
from pyjeeves import logging
|
|
logger = logging.getLogger("PyJeeves." + __name__)
|
|
|
|
|
|
# Relocate Jeeves modules to separate folder and let a "master" module handle imports, and setup.
|
|
class Company():
|
|
"""Handles companies in Jeeves"""
|
|
|
|
@staticmethod
|
|
def get(ftg_nr):
|
|
""" Query an article by number """
|
|
try:
|
|
return db.raw.query(CompanyModel).filter_by(
|
|
FtgNr=ftg_nr
|
|
).one()
|
|
except NoResultFound:
|
|
raise KeyError
|
|
|
|
@staticmethod
|
|
def get_all_active_customers():
|
|
cust = db.raw.query(CustomerModel).filter(and_(CustomerModel.Makulerad == 0)).all()
|
|
return [c.CompanyModel for c in cust]
|
|
|
|
@staticmethod
|
|
def get_customer_numbers(category_list=[10], class_list=[], filter_inactive=True):
|
|
category_in = CustomerModel.kundkategorikod.in_(category_list) if category_list else and_()
|
|
class_in = CustomerModel.kundklass.in_(class_list) if class_list else and_()
|
|
inactive = and_(CustomerModel.Makulerad == 0) if filter_inactive else and_()
|
|
cust = db.raw.query(CustomerModel).options(
|
|
Load(CustomerModel).noload('*')).filter(
|
|
and_(category_in, class_in, inactive)).all()
|
|
return [c.FtgNr for c in cust]
|
|
|
|
@staticmethod
|
|
def get_list(ftg_nr=[], filter_=and_(CustomerModel.Makulerad == 0), offset=0, limit=100):
|
|
ftg_filter = and_()
|
|
if ftg_nr:
|
|
ftg_filter = CompanyModel.FtgNr.in_(ftg_nr)
|
|
|
|
return db.raw.query(CompanyModel).join(CustomerModel).filter(
|
|
and_(ftg_filter, filter_)).order_by(
|
|
CompanyModel.FtgNr.desc()).offset(offset).limit(limit).all()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# print([column.key for column in CompanyModel.__table__.columns])
|
|
|
|
logger.info("Starting TEST")
|
|
# session = RawSession()
|
|
|
|
logger.info("Testing gettings a company")
|
|
# c1 = session.query(CompanyModel).filter_by(FtgNr="179580").first()
|
|
# print(CompanyModel)
|
|
|
|
# print(CompanyModel.get_list(['406569', '179580', '2440070', '179584']))
|
|
from pprint import pprint
|
|
pprint(CompanyModel.get('179584').to_dict())
|
|
|
|
# c1 = CompanyModel.query.filter_by(FtgNr="406569").first()
|
|
# print(c1)
|
|
# logger.info(c1.json)
|
|
|
|
# print(
|
|
# len(CompanyModel.get_all_active_customers())
|
|
# )
|
|
|
|
# print(CompanyModel.get_all_active_customers()[0].CompanyModel)
|