KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use create_schema method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

test_schema.py

Source:test_schema.py

...148        self.session = Session()149        self.url = 'http://example.org/schema'150        self.pointer = ''151        self.href = '#'.join((self.url, self.pointer))152    def create_schema(self, href):153        schema = deepcopy(SCHEMA)154        return Schema(155            href, raw_schema=schema, session=self.session)156    def assertValidRefs(self, schema, url=None, pointer=None):157        url = url or self.url158        pointer = pointer or self.pointer159        href = '#'.join((url, pointer))160        self.assertEqual(schema.url, url)161        self.assertEqual(schema.pointer, pointer)162        self.assertEqual(schema.href, href)163    def test_splits_url_and_pointer_on_init(self):164        schema = self.create_schema(self.href)165        self.assertValidRefs(schema)166    def test_fixes_href_without_pointer(self):167        schema = self.create_schema(self.url)168        self.assertValidRefs(schema)169    def test_fixes_empty_pointer(self):170        schema = self.create_schema(self.url + '#')171        self.assertValidRefs(schema)172    def test_accepts_pointer(self):173        pointer = '/properties/name'174        schema = self.create_schema(self.href + pointer)175        self.assertValidRefs(schema, pointer=pointer)176    def test_returns_root_schema_data_from_empty_pointer(self):177        schema = self.create_schema(self.href)178        self.assertEqual(schema['title'], SCHEMA['title'])179    def test_returns_schema_data_from_pointer(self):180        pointer = '/properties/name'181        schema = self.create_schema(self.href + pointer)182        self.assertEqual(schema, SCHEMA['properties']['name'])183    def test_resolves_local_pointer_on_objects(self):184        schema = self.create_schema(self.href)185        self.assertEqual(schema['properties']['pointer'], SCHEMA['pointer'])186    def test_resolves_local_pointer_on_arrays(self):187        schema = self.create_schema(self.href)188        pointers = schema['properties']['pointers']['items']['oneOf']189        expected = [SCHEMA['pointer'], SCHEMA['pointer2']]190        self.assertEqual(pointers, expected)191    def test_keeps_context_between_refs(self):192        schema = self.create_schema(self.href)193        self.assertEqual(schema['properties']['repointer'], SCHEMA['pointer'])194    def test_resolves_external_ref_with_lazy_schema(self):195        schema = self.create_schema(self.href)196        external = schema['properties']['external']197        self.assertIsInstance(external, LazySchema)198        self.assertEqual(199            external.href, SCHEMA['properties']['external']['$ref'])200    def test_resolves_self_reference(self):201        schema = self.create_schema(self.href)202        self_ref = schema['properties']['self']203        self.assertEqual(204            self_ref['title'], SCHEMA['title'])205    def test_is_instance_of_dict(self):206        schema = self.create_schema(self.href)207        self.assertIsInstance(schema, dict)208    def test_is_instance_of_schema(self):209        schema = self.create_schema(self.href)210        self.assertIsInstance(schema, Schema)211class LazySchemaPointerTestCase(BaseLazySchemaTestCase):212    HREF = '/schema#/properties/name'213    def test_applies_pointer_to_loaded_schema(self):214        self.assertEqual(self.schema, SCHEMA['properties']['name'])215class SchemaStoreMixinTestCase(object):216    SCHEMA_URL = 'http://a.com/schema'217    def setUp(self):218        self.session = Session()219    def test_reuses_schema_for_same_href(self):220        schema1 = self.create_schema(self.SCHEMA_URL)221        schema2 = self.create_schema(self.SCHEMA_URL)222        self.assertIs(schema1, schema2)223    def test_doesnt_reuse_schema_with_empty_url(self):224        schema1 = self.create_schema('')225        schema2 = self.create_schema('')226        self.assertIs(schema1, schema2)227    def test_reuses_schema_for_different_pointers(self):228        schema1 = self.create_schema(self.SCHEMA_URL)229        schema2 = self.create_schema(230            self.SCHEMA_URL + '#/pointer')231        self.assertIs(schema1, schema2)232    def test_does_not_reuse_schema_on_different_sessions(self):233        other_session = Session()234        schema1 = self.create_schema(self.SCHEMA_URL)235        schema2 = self.create_schema(236            self.SCHEMA_URL, session=other_session)237        self.assertIsNot(schema1, schema2)238class SchemaStoreTestCase(SchemaStoreMixinTestCase, TestCase):239    def create_schema(self, url, session=None):240        if session is None:241            session = self.session242        return Schema(self.SCHEMA_URL, raw_schema={}, session=session)243class LazySchemaStoreTestCase(SchemaStoreMixinTestCase, TestCase):244    def create_schema(self, url, session=None):245        if session is None:246            session = self.session...

router.py

Source:router.py

12from fastapi import Depends, HTTPException3from fastapi_crudrouter import SQLAlchemyCRUDRouter4from fastapi.security import OAuth2PasswordRequestForm5from app.db import Base, get_db6from app.security import (UserInDB, abm_users_db, fake_hash_password,7                          token_auth1)8from app.auth import get_current_user, authenticate_user91011''' SORTED IMPORTS '''1213from app.tables.accounts import *14from app.tables.bayerpersona import *15from app.tables.bc_bp import *16from app.tables.bc_icp import *17from app.tables.bc_pers import *18from app.tables.bj_icp import *19from app.tables.buyersjourney import *20from app.tables.buyingcomitee import *21from app.tables.camp_cont import *22from app.tables.camp_kpi import *23from app.tables.camp_pers import *24from app.tables.camp_stgs import *25from app.tables.campaign_goals import *26from app.tables.campaign import *27from app.tables.cg_camp import *28from app.tables.companies import *29from app.tables.cont_type import *30from app.tables.content import *31from app.tables.distr_channels import *32from app.tables.icp_ind import *33from app.tables.icp_tier import *34from app.tables.icp import * 35from app.tables.industries import *36from app.tables.jobtitle import *37from app.tables.kpi import *38from app.tables.personas import *39from app.tables.rolebp import *40from app.tables.servprod import *41from app.tables.servprodcomp import *42from app.tables.stages import *43from app.tables.tiers import *444546''' Routers MODELO4748router = SQLAlchemyCRUDRouter(49    schema=,50    create_schema=,51    db_model=,52    dependencies=[Depends(token_auth1)],53    db=get_db,54    prefix=''55)  5657'''5859'''         CRUD '''606162routerAccounts = SQLAlchemyCRUDRouter(63    schema=accounts,64    create_schema=accountsCreate,65    db_model=accountsModel,66    dependencies=[Depends(authenticate_user)],67    db=get_db,68    prefix='accounts'69)707172routerBayer = SQLAlchemyCRUDRouter(73    schema=Bayerpersona,74    create_schema=BayerpersonaCreate,75    db_model=BayerpersonaModel,76    # dependencies=[Depends(token_auth1)],77    db=get_db,78    prefix='bayerpersona'79)8081routerBc_bp = SQLAlchemyCRUDRouter(82    schema=bc_bp,83    create_schema=bc_bpCreate,84    db_model=bc_bpModel,85    # dependencies=[Depends(token_auth1)],86    db=get_db,87    prefix='bc_bp'88)8990routerBc_icp = SQLAlchemyCRUDRouter(91    schema=Bc_icp,92    create_schema=Bc_icpCreate,93    db_model=Bc_icp_Model,94    # dependencies=[Depends(token_auth1)],95    db=get_db,96    prefix='bc_icp'97)9899routerbc_pers = SQLAlchemyCRUDRouter(100    schema=bc_pers,101    create_schema=bc_persCreate,102    db_model=bc_persModel,103    # dependencies=[Depends(token_auth1)],104    db=get_db,105    prefix='bc_pers'106)107108routerbj_icp = SQLAlchemyCRUDRouter(109    schema=bj_icp,110    create_schema=bj_icpCreate,111    db_model=bj_icpModel,112    # dependencies=[Depends(token_auth1)],113    db=get_db,114    prefix='bj_icp'115)116117routerbuyersjourney = SQLAlchemyCRUDRouter(118    schema=buyersjourney,119    create_schema=buyersjourneyCreate,120    db_model=buyersjourneyModel,121    # dependencies=[Depends(token_auth1)],122    db=get_db,123    prefix='buyersjourney'124)125126127routerBuyingcomitee = SQLAlchemyCRUDRouter(128    schema=buyingcomitee,129    create_schema=buyingcomiteeCreate,130    db_model=buyingcomiteeModel,131    # dependencies=[Depends(token_auth1)],132    db=get_db,133    prefix='buyingcomitee'134)135136routerCamp_cont = SQLAlchemyCRUDRouter(137    schema = Camp_cont,138    create_schema = Camp_contCreate,139    db_model = Camp_cont_Model,140    # dependencies=[Depends(token_auth1)],141    db = get_db,142    prefix = "camp_cont"143)144145routerCamp_KPI = SQLAlchemyCRUDRouter(146    schema=Camp_KPI,147    create_schema=Camp_KPICreate,148    db_model=Camp_KPIModel,149    # dependencies=[Depends(token_auth1)],150    db=get_db,151    prefix='camp_kpi'152)153154routerCamp_pers = SQLAlchemyCRUDRouter(155    schema=camp_pers,156    create_schema=camp_persCreate,157    db_model=camp_persModel,158    # dependencies=[Depends(token_auth1)],159    db=get_db,160    prefix='camp_pers'161)162163164routerCamp_stgs = SQLAlchemyCRUDRouter(165    schema=Camp_stgs,166    create_schema=Camp_stgsCreate,167    db_model=Camp_stgs_Model,168    # dependencies=[Depends(token_auth1)],169    db=get_db,170    prefix='camp_stgs'171)172173routerCampaign_goals = SQLAlchemyCRUDRouter(174    schema = Campaign_goals,175    create_schema = Campaign_goalsCreate,176    db_model = Campaign_goals_Model,177    # dependencies=[Depends(token_auth1)],178    db = get_db,179    prefix = "campaing_goals"180)181182routerCampaign = SQLAlchemyCRUDRouter(183    schema=Campaign,184    create_schema=CampaignCreate,185    db_model=CampaignModel,186    # dependencies=[Depends(token_auth1)],187    db=get_db,188    prefix='campaigns'189)190191192routerCg_camp = SQLAlchemyCRUDRouter(193    schema = Cg_camp,194    create_schema = Cg_campCreate,195    db_model = Cg_camp_Model,196    # dependencies=[Depends(token_auth1)],197    db = get_db,198    prefix = "cg_camp"199)200201202routercompanies = SQLAlchemyCRUDRouter(203    schema = Companies,204    create_schema = CompaniesCreate,205    db_model = CompaniesModel,206    # dependencies=[Depends(token_auth1)],207    db = get_db,208    prefix = "Companies"209)210211routerCont_type = SQLAlchemyCRUDRouter(212    schema=cont_type,213    create_schema=cont_typeCreate,214    db_model=cont_typeModel,215    # dependencies=[Depends(token_auth1)],216    db=get_db,217    prefix='cont_type'218)219220routerContent = SQLAlchemyCRUDRouter(221    schema=content,222    create_schema=contentCreate,223    db_model=contentModel,224    # dependencies=[Depends(token_auth1)],225    db=get_db,226    prefix='content'227)228229routerdistr_channels = SQLAlchemyCRUDRouter(230    schema=distr_channels,231    create_schema=distr_channelsCreate,232    db_model=distr_channelsModel,233    # dependencies=[Depends(token_auth1)],234    db=get_db,235    prefix='distr_channels'236)237238routerIcp_ind = SQLAlchemyCRUDRouter(239    schema = Icp_ind,240    create_schema = Icp_indCreate,241    db_model = Icp_ind_Model,242    #dependencies=[Depends(token_auth1)],243    db = get_db,244    prefix = "Icp_ind"245)246247248routerIcp = SQLAlchemyCRUDRouter(249    schema = Icp,250    create_schema = IcpCreate,251    db_model = Icp_Model,252    #dependencies=[Depends(token_auth1)],253    db = get_db,254    prefix = "Icp"255)256257routerIndustries = SQLAlchemyCRUDRouter(258    schema=Industries,259    create_schema=IndustriesCreate,260    db_model=IndustriesModel,261    #dependencies=[Depends(token_auth1)],262    db=get_db,263    prefix='industries'264)265266routerIcp_tier = SQLAlchemyCRUDRouter(267    schema=icp_tier,268    create_schema=icp_tierCreate,269    db_model=icp_tierModel,270    #dependencies=[Depends(token_auth1)],271    db=get_db,272    prefix='icp_tier'273) 274275routerJobtitle = SQLAlchemyCRUDRouter(276    schema=jobtitle,277    create_schema=jobtitleCreate,278    db_model=jobtitleModel,279    #dependencies=[Depends(token_auth1)],280    db=get_db,281    prefix='jobtitle'282)283284routerKPI = SQLAlchemyCRUDRouter(285    schema=KPI,286    create_schema=KPICreate,287    db_model=KPIModel,288    #dependencies=[Depends(token_auth1)],289    db=get_db,290    prefix='kpi'291)292293routerPersonas = SQLAlchemyCRUDRouter(294    schema=personas,295    create_schema=personasCreate,296    db_model=personasModel,297    #dependencies=[Depends(token_auth1)],298    db=get_db,299    prefix='personas'300)301302routerRole_bp = SQLAlchemyCRUDRouter(303    schema=Role_bp,304    create_schema=RolebpCreate,305    db_model=Role_bpModel,306    #dependencies=[Depends(token_auth1)],307    db=get_db,308    prefix='Role_bp'309)310311312routerServprod = SQLAlchemyCRUDRouter(313    schema=Servprod,314    create_schema=ServprodCreate,315    db_model=ServprodModel,316    #dependencies=[Depends(token_auth1)],317    db=get_db,318    prefix='servprod'319)320321routerServprodcomp = SQLAlchemyCRUDRouter(322    schema=Servprodcomp,323    create_schema=Servprodcomp_Create,324    db_model=ServprodcompModel,325    #dependencies=[Depends(token_auth1)],326    db=get_db,327    prefix='servprodcomp'328)329330routerStages = SQLAlchemyCRUDRouter(331    schema=stages,332    create_schema=stagesCreate,333    db_model=stagesModel,334    #dependencies=[Depends(token_auth1)],335    db=get_db,336    prefix='stages'337)338339routertiers = SQLAlchemyCRUDRouter(340    schema=tiers,341    create_schema=tiersCreate,342    db_model=tiersModel,343    #dependencies=[Depends(token_auth1)],344    db=get_db,345    prefix='tiers'
...

routing.py

Source:routing.py

1from api.Schemas import *2from api.models import models3from api.Services import get_db4from fastapi_crudrouter import SQLAlchemyCRUDRouter5router_manager = SQLAlchemyCRUDRouter(6    schema= RoleScheme,7    create_schema=RoleSchemeCreate,8    delete_all_route=False,9    db_model=models.ManagerPerson,10    db=get_db,11    tags=["managers"],12    prefix='/api/managers',13)14router_customer = SQLAlchemyCRUDRouter(15    schema= CustomerScheme,16    create_schema=RoleSchemeCreate,17    update_schema=CustomerScheme,18    delete_all_route=False,19    db_model=models.Customer,20    db=get_db,21    tags=["customers"],22    prefix='/api/customers',23)24router_admins = SQLAlchemyCRUDRouter(25    schema= AdminScheme,26    create_schema=RoleSchemeCreate,27    update_schema=AdminScheme,28    delete_all_route=False,29    db_model=models.Administator,30    db=get_db,31    tags=["admins"],32        prefix='/api/administrators',33)34router_users = SQLAlchemyCRUDRouter(35    schema= User,36    create_schema=UserCreate,37    update_schema=User,38    delete_all_route=False,39    db_model=models.User,40    db=get_db,41    tags=["user"],42    prefix='/api/user',43)44router_tables = SQLAlchemyCRUDRouter(45    schema= TableScheme,46    create_schema=TableSchemeCreate,47    update_schema=TableScheme,48    delete_all_route=False,49    db_model=models.Table,50    db=get_db,51    tags=["tables"],52    prefix='/api/tables',53)54router_dishes = SQLAlchemyCRUDRouter(55    schema= DishScheme,56    create_schema=DishSchemeCreate,57    update_schema=DishScheme,58    delete_all_route=False,59    db_model=models.Dish,60    db=get_db,61    tags=["dishes"],62    prefix='/api/dishes',63)64router_orders = SQLAlchemyCRUDRouter(65    schema= OrderScheme,66    create_schema=OrderSchemeCreate,67    update_schema=OrderScheme,68    delete_all_route=False,69    db_model=models.Order,70    db=get_db,71    tags=["orders"],72    prefix='/api/orders',73)74router_comments = SQLAlchemyCRUDRouter(75    schema=CommentsScheme,76    create_schema=CommentsScheme,77    update_schema=CommentsSchemeUpdate,78    delete_all_route=False,79    db_model=models.Comment,80    db=get_db,81    tags=["comments"],82    prefix='/api/comments',83)84router_news = SQLAlchemyCRUDRouter(85    schema=NewsScheme,86    create_schema=NewsScheme,87    update_schema=NewsScheme,88    delete_all_route=False,89    db_model=models.News,90    db=get_db,91    tags=["news"],92    prefix='/api/news',...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.