from sqlalchemy.orm import Session from sqlalchemy import text from kv.database import get_table from kv import schemas def exec_procedure(session, proc_name, params, database: str = None): sql_params = ",".join([f"@{name}='{value}'" if type(value) in [str] else f"@{name}={value}" for name, value in params.items()]) dbstr = "" if database: dbstr = f"[{database}]." sql_string = """ DECLARE @return_value int; EXEC @return_value = {dbstr}[dbo].[{proc_name}] {params}; SELECT 'Return Value' = @return_value; """.format(dbstr=dbstr, proc_name=proc_name, params=sql_params) return session.execute(sql_string).fetchall() def exec_procedure_wo_result(session, proc_name, params, database: str = None): sql_params = ",".join([f"@{name}='{value}'" if type(value) in [str] else f"@{name}={value}" for name, value in params.items()]) dbstr = "" if database: dbstr = f"[{database}]." sql_string = """ EXEC {dbstr}[dbo].[{proc_name}] {params} """.format(dbstr=dbstr, proc_name=proc_name, params=sql_params) session.execute(sql_string) session.commit() """PersonalAccountViewSet""" class PersonalAccountViewSet: def get_turn_over(db: Session, data: schemas.PersonalAccountInit = None): return exec_procedure(db, 'uspGetPersonalAccountTurnOver', **data) def get_details(db: Session, id: int = None): table = get_table("PersonalAccount") return db.query(table).filter(table.c["ID"] == id).first() def get_services(db: Session, IDPersonalAccount: int = None): return exec_procedure(db, 'uspGetPersonalAccountServices', {'IDPersonalAccount': IDPersonalAccount}) def get_debt(db: Session, data: schemas.PersonalAccount = None): return exec_procedure(db, 'uspGetPersonalAccountDebt', **data) def get_debts(db: Session, data: schemas.PersonalAccountType = None): return exec_procedure(db, 'uspGetPersonalAccountDebts', **data) def get_financial_account(db: Session, data: schemas.PersonalAccountReportTOFinancialInit = None): return exec_procedure(db, 'uspGetPersonalAccountReportTOFinancialAccount', **data) """AddressInfoViewSet""" class AddressInfoViewSet: """get""" def get_details(db: Session, id: int = None): table = get_table("RefAddresses", "General") return db.query(table).filter(table.c["ID"] == id).first() def get_personal_accounts(db: Session, IDCity: int = None): return exec_procedure(db, 'uspGetPersonalAccountsList', {'IDCity': IDCity}) def get_addresses(db: Session, IDCity: int = None): return exec_procedure(db, 'uspGetAddressesByCity', {'IDCity': IDCity}) def get_address_personal_accounts(db: Session, IDAddress: int = None): return exec_procedure(db, 'uspGetAddressPersonalAccounts', {'IDAddress': IDAddress}) def retrieve(db: Session, id: int = None): return exec_procedure(db, 'uspGetAddressAndObjectDetails', {'ID': id}) def get_metering_devices(db: Session, IDAddress: int = None): return exec_procedure(db, 'uspGetAddressMeteringDevices', {'IDAddress': IDAddress}) """post""" def edit(db: Session, data: schemas.EditAdressInit = None): try: exec_procedure_wo_result(db, 'uspEditAddress', **data) return {'msg': 'success'} except: return {'msg': 'error'} def edit_personal_account(db: Session, data: schemas.PersonalAccountDetailsSerializer = None): try: exec_procedure_wo_result(db, 'uspEditPersonalAccount', **data) return {'msg': 'success'} except: return {'msg': 'error'} def edit_personal_account_address(db: Session, data: schemas.EditPersonalAccountAddressInit = None): try: exec_procedure_wo_result( db, 'uspEditPersonalAccountAddress', **data) return {'msg': 'success'} except: return {'msg': 'error'} """ObjectViewSet""" class ObjectViewSet: """get""" def get_details(db: Session, IDObject: str = None): print(IDObject) return exec_procedure(db, 'uspGetObjectDetails', {'IDObject': IDObject}) def get_municipals(db: Session, IDCity: int = None): return exec_procedure(db, 'uspGetObjectMunicipals', {'IDCity': IDCity}) def get_services(db: Session, IDObject: str = None): return exec_procedure(db, 'uspGetObjectMunicipals', {'IDObject': IDObject}) def get_addresses(db: Session, IDObject: str = None): return exec_procedure(db, 'uspGetObjectAddresses', {'IDObject': IDObject}) def get_metering_devices(db: Session, IDObject: str = None): return exec_procedure(db, 'uspGetObjectMeteringDevices', {'IDObject': IDObject}) """post""" def edit(db: Session, data: schemas.ObjectDetailsSerializer = None): try: exec_procedure_wo_result(db, 'uspEditObject', **data) return {'msg': 'success'} except: return {'msg': 'error'} def edit_service(db: Session, data: schemas.EditObjectServiceInit = None): try: exec_procedure_wo_result(db, 'uspEditObject', **data) return {'msg': 'success'} except: return {'msg': 'error'} def edit_temp_metering_device_address(db: Session, data: schemas.EditTempMeteringDeviceAddressInit = None): try: exec_procedure_wo_result( db, 'uspEditTempMeteringDeviceAddress', **data, database='General') return {'msg': 'success'} except: return {'msg': 'error'} def add_object(db: Session, data: schemas.AddObjectInit = None): try: exec_procedure_wo_result(db, 'uspAddObject', **data) return {'msg': 'success'} except: return {'msg': 'error'} """TurnOverViewSet""" class TurnOverViewSet: """get""" """post""" def get_to_archive(db: Session, data: schemas.ArchiveInit = None): return exec_procedure(db, 'uspGetTOArchive', **data) def make_storno(db: Session, data: schemas.StronoInit = None): try: exec_procedure_wo_result(db, 'uspStorno', **data) return {'msg': 'success'} except: return {'msg': 'error'} """PaymentViewSet""" class PaymentViewSet: """get""" def get_money_types(db: Session): return db.query(get_table('RefMoneyTypes', 'General')).all() def get_worker_types(db: Session): return db.query(get_table('RefWorkerTypes', 'General')).all() def get_payer_category_types(db: Session): return db.query(get_table('RefPayerCategoryTypes', 'General')).all() def get_services_additional(db: Session): uslugi_tb = get_table('uslugi', 'spr') data_query = db.query(uslugi_tb).filter(uslugi_tb.c["vid_usl"] == 2).order_by( uslugi_tb.c["name"]).with_entities(uslugi_tb.c["kod"].label("ID"), uslugi_tb.c["name"].label("Name")) return data_query.all() """post""" def get_receipt_calculate(db: Session, data: schemas.RecieptCalculateInit): return exec_procedure(db, 'uspGetReceiptCalculate', **data) def save_receipt(db: Session, data: schemas.ReceiptSaveInit = None): try: exec_procedure_wo_result(db, 'uspSaveReceipt', **data) return {'msg': 'success'} except: return {'msg': 'error'} def repayment_info(db: Session, data: schemas.RepaymentInfoInit): return exec_procedure(db, 'uspGetRepaymentInfo', **data) def repayment(db: Session, data: schemas.RepaymentInit): return exec_procedure(db, 'uspRepayment', **data) """ReportViewSet""" class ReportViewSet: """get""" """post""" def repayment(db: Session, data: schemas.RentRegisterNoticeInit): return exec_procedure(db, 'uspGetRentRegisterNotices', **data) """ReportFilterViewSet""" class ReportFilterViewSet: """get""" """post""" def get_filter_types(db: Session, data: schemas.FilterTypeInit): return exec_procedure(db, 'uspGetFilterTypes', data.dict()) def get_filter_list(db: Session, data: schemas.FilterListInit): return exec_procedure(db, 'uspGetFilterList', data.dict()) def set_report_filter_list(db: Session, data: schemas.ReportFilterSetInit): return exec_procedure_wo_result(db, 'uspSetFilterList', data.dict()) """AccrualViewSet""" class AccrualViewSet: """get""" def get_municipals(db: Session, data: schemas.AccountInit): return exec_procedure(db, 'uspCalculateAccrued', data.dict()) """post""" """WithdrawingViewSet""" class WithdrawingViewSet: """get""" def get_withdrawing_types(db: Session): return db.query(get_table('RefWithdrawingTypes', 'General')).all() """post""" def set_report_filter_list(db: Session, data: schemas.WithdrawingCheckInit): return exec_procedure(db, 'uspGetWithdrawingCheck', **data) def get_withdrawing_temp(db: Session, data: schemas.WithdrawingTempInit): return exec_procedure(db, 'uspGetWithdrawingTemp', **data) def save_withdrawing(db: Session, data: schemas.WithdrawingSave = None): try: exec_procedure_wo_result(db, 'uspSaveWithdrawing', **data) return {'msg': 'success'} except: return {'msg': 'error'} """ReferenceViewSet""" class ReferenceViewSet: """get""" def get_personal_accounts(db: Session): tb = get_table('PersonalAccount') return db.query(tb).with_entities(tb.c["ID"]).all() def get_objects(db: Session, data: schemas.CityInit): return exec_procedure(db, 'uspGetObjects', data.dict()) def get_services(db: Session): tb = get_table('uslugi', 'spr') query = db.query(tb).with_entities( tb.c["kod"].label("ID"), tb.c["name"].label("Name")) return query.all() def get_provider_types(db: Session): tb = get_table('RefProviderTypes', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_providers(db: Session): tb = get_table('RefProviders', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_systems_water(db: Session): tb = get_table('RefSystemsWater') return db.query(tb).all() def get_systems_water_period_types(db: Session): tb = get_table('RefSystemsWaterPeriodTypes') return db.query(tb).all() def get_provider_contracts(db: Session, data: schemas.PAInit): return exec_procedure(db, 'uspGetProviderContracts', data.dict()) def get_providers_utility_services(db: Session, data: schemas.CityInit): tb = get_table('RefProvidersUtilityServices', 'General') data_query = db.query(tb).filter(tb.c["IDCity"] == data.IDCity).with_entities( tb.c["ID"], tb.c["ShortName"].label("Name")) return data_query.all() def get_management_companies(db: Session): tb = get_table('RefManagementCompany', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_boilers_by_city(db: Session, data: schemas.CityInit): tb = get_table('vBoilers', 'General') data_query = db.query(tb).filter(tb.c["id_city"] == data.IDCity).with_entities( tb.c["ID"], tb.c["Name"]) return data_query.all() def get_streets_by_city(db: Session, data: schemas.CityInit): tb = get_table('vStreets', 'General') data_query = db.query(tb).filter(tb.c["id_city"] == data.IDCity).with_entities( tb.c["ID"], tb.c["Name"]) return data_query.all() def get_metering_devices_services(db: Session): qu = text( "select kod ID, case when kod=3 then 'Горячая вода' else sname end Name from spr..uslugi where uchet=1 order by kod") return db.execute(qu).fetchall() def get_owhership_types(db: Session): tb = get_table('RefObjectOwnershipTypes', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_object_improvement_degree_types(db: Session): tb = get_table('RefObjectImprovementDegreeTypes', 'General') return db.query(tb).all() def get_object_gas_types(db: Session): tb = get_table('RefObjectGasTypes', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_object_heating_types(db: Session): tb = get_table('RefObjectHeatingTypes', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_object_wiring_types(db: Session): tb = get_table('RefObjectWiringTypes', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_object_overlapping_types(db: Session): tb = get_table('RefObjectOverlappingTypes', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_object_wall_material_types(db: Session): tb = get_table('RefObjectWallMaterialTypes', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_object_roof_types(db: Session): tb = get_table('RefObjectRoofTypes', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_object_foundation_types(db: Session): tb = get_table('RefObjectFoundationTypes', 'General') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_tariffs_population(db: Session): tb = get_table('TariffPopulation') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_courts_types(db: Session): tb = get_table('RefCourtsTypes') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() def get_address_types(db: Session): tb = get_table('RefAddressType') return db.query(tb).with_entities(tb.c["ID"], tb.c["Name"]).all() """post""" def get_advanced_search(db: Session, data: schemas.AdvancedSearchInit): return exec_procedure(db, 'uspGetAdvancedSearch', **data) def get_metering_devices_types(db: Session, data: schemas.ServiceIDInit): tb = get_table('RefMeteringDevicesTypes', 'General') data_query = db.query(tb).filter(tb.c["IDService"] == data.IDService).order_by( tb.c["Name"]).with_entities(tb.c["ID"], tb.c["Name"]) return data_query.all() def get_object_metering_devices_types(db: Session, data: schemas.ObjectMeteringDevicesTypesInit): return exec_procedure(db, 'uspGetMeteringDeviceTypes', params=data, database='General') def get_object_metering_device_addresses(db: Session, data: schemas.ObjectMeteringDeviceAddressInit): return exec_procedure(db, 'uspGetMeteringDeviceTypes', params=data, database='General') def get_object_water_system_volumes(db: Session, data: schemas.ObjectWaterSystemVolumesInit): return exec_procedure(db, 'uspGetObjectWaterSystemVolumes', **data) def get_tariff_population(db: Session, data: schemas.TariffsPopulationInit): return exec_procedure(db, 'uspGetTariffsPopulation', **data) def get_object_tariff_provider(db: Session, data: schemas.TariffProviderInit): return exec_procedure(db, 'uspGetObjectTariffProvider', **data) def edit_metering_device(db: Session, data: schemas.MeteringDeviceInit): return exec_procedure_wo_result(db, 'uspEditMeteringDevice', **data)