fix relation target for foreign tree api building

Author: LuisLuii

src/fastapi_quickcrud/crud_router.py

@@ -152,7 +152,8 @@ async def async_runner(f):
     foreign_table_mapping = {db_model.__tablename__: db_model}
     if foreign_include:
         for i in foreign_include:
-            foreign_table_mapping[i.__tablename__] = i
+            model , _= convert_table_to_model(i)
+            foreign_table_mapping[model.__tablename__] = i
     crud_service = query_service(model=db_model, async_mode=async_mode, foreign_table_mapping=foreign_table_mapping)
     # else:
     #     crud_service = SQLAlchemyPostgreQueryService(model=db_model, async_mode=async_mode)

src/fastapi_quickcrud/misc/covert_model.py

@@ -0,0 +1,24 @@
+from sqlalchemy.ext.declarative import declarative_base
+
+from sqlalchemy.sql.schema import Table
+
+
+def convert_table_to_model(db_model):
+    NO_PRIMARY_KEY = False
+    if not isinstance(db_model, Table):
+        return db_model, NO_PRIMARY_KEY
+    db_name = str(db_model.fullname)
+    table_dict = {'__table__': db_model,
+                  '__tablename__': db_name}
+
+    if not db_model.primary_key:
+        table_dict['__mapper_args__'] = {
+            "primary_key": [i for i in db_model._columns]
+        }
+        NO_PRIMARY_KEY = True
+
+    for i in db_model.c:
+        col, = i.expression.base_columns
+        table_dict[str(i.key)] = col
+
+    return type(f'{db_name}DeclarativeBaseClass', (declarative_base(),), table_dict), NO_PRIMARY_KEY

src/fastapi_quickcrud/misc/schema_builder.py

@@ -27,6 +27,7 @@
 from sqlalchemy.orm import declarative_base
 from strenum import StrEnum
 
+from .covert_model import convert_table_to_model
 from .exceptions import (SchemaException,
                          ColumnTypeNotSupportedException)
 from .type import (MatchingPatternInStringBase,
@@ -182,6 +183,7 @@ def __foreign_mapper_builder(self):
         if self.exclude_primary_key:
             return foreign_mapper
         for db_model in self.foreign_include:
+            db_model, NO_PRIMARY_KEY = convert_table_to_model(db_model)
             tmp = {}
             table_name = self.__get_table_name(db_model)
             tmp["model"] = db_model
@@ -489,12 +491,15 @@ def _extra_relation_level(self, model=None, processed_table=None) -> Dict[str, T
         mapper = inspect(model)
         relation_level = []
         for r in mapper.relationships:
-            relation_table = r.key
-            if relation_table and relation_table not in processed_table and relation_table in self.foreign_mapper:
+
+            target_table = r.target
+            target_model, _ = convert_table_to_model(target_table)
+            target_table_name = target_model.__tablename__
+            if target_table_name and target_table_name not in processed_table and target_table_name in self.foreign_mapper:
                 processed_table.append(str(mapper.local_table))
-                if self.foreign_mapper[relation_table]["db_name"] not in relation_level:
-                    relation_level.append(self.foreign_mapper[relation_table]["db_name"])
-                relation_level += self._extra_relation_level(self.foreign_mapper[relation_table]["db_model"],
+                if self.foreign_mapper[target_table_name]["db_name"] not in relation_level:
+                    relation_level.append(self.foreign_mapper[target_table_name]["db_name"])
+                relation_level += self._extra_relation_level(self.foreign_mapper[target_table_name]["db_model"],
                                                              processed_table=processed_table
                                                              )
         return relation_level
@@ -505,7 +510,6 @@ def _extra_foreign_table_from_declarative_base(self, model) -> Dict[str, Table]:
         reference_mapper = {}
         for r in mapper.relationships:
             local, = r.local_columns
-            relation_table = r.key
             local = mapper.get_property_by_column(local).expression
             local_table = str(local).split('.')[0]
             local_column = str(local).split('.')[1]

src/fastapi_quickcrud/misc/utils.py

@@ -9,6 +9,8 @@
     BinaryExpression
 
 from sqlalchemy.sql.schema import Table
+
+from .covert_model import convert_table_to_model
 from .crud_model import RequestResponseModel, CRUDModel
 from .exceptions import QueryOperatorNotFound, PrimaryMissing, UnknownColumn
 from .schema_builder import ApiParameterSchemaBuilder
@@ -337,27 +339,6 @@ def group_by_foreign_key(item):
     return response_list
 
 
-def convert_table_to_model(db_model):
-    NO_PRIMARY_KEY = False
-    if not isinstance(db_model, Table):
-        return db_model, NO_PRIMARY_KEY
-    db_name = str(db_model.fullname)
-    table_dict = {'__table__': db_model,
-                  '__tablename__': db_name}
-
-    if not db_model.primary_key:
-        table_dict['__mapper_args__'] = {
-            "primary_key": [i for i in db_model._columns]
-        }
-        NO_PRIMARY_KEY = True
-
-    for i in db_model.c:
-        col, = i.expression.base_columns
-        table_dict[str(i.key)] = col
-
-    return type(f'{db_name}DeclarativeBaseClass', (declarative_base(),), table_dict), NO_PRIMARY_KEY
-
-
 def path_query_builder(params, model) -> List[Union[BinaryExpression]]:
     query = []
     if not params:

src/fastapi_quickcrud/sample_test.py

@@ -0,0 +1,68 @@
+from fastapi import FastAPI
+from fastapi_quickcrud import crud_router_builder
+from sqlalchemy import *
+from sqlalchemy.orm import *
+from fastapi_quickcrud.crud_router import generic_sql_crud_router_builder
+
+Base = declarative_base()
+
+
+class Account(Base):
+    __tablename__ = "account"
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    post = relationship("BlogPost", back_populates="account_f")
+
+
+class BlogPost(Base):
+    __tablename__ = "blog_post"
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    account_id = Column(Integer, ForeignKey("account.id"), nullable=False)
+    account_f = relationship("Account", back_populates="post")
+    blog = relationship("BlogComment", back_populates="post")
+
+
+class BlogComment(Base):
+    __tablename__ = "blog_comment"
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    blog_id = Column(Integer, ForeignKey("blog_post.id"), nullable=False)
+    post = relationship("BlogPost", back_populates="blog")
+
+
+crud_route_parent = crud_router_builder(
+    db_model=Account,
+    prefix="/account",
+    tags=["account"],
+    foreign_include=[BlogComment, BlogPost]
+
+)
+print("BlogPost222")
+
+crud_route_child1 = generic_sql_crud_router_builder(
+    db_model=BlogPost,
+    prefix="/blog_post",
+    tags=["blog_post"],
+    foreign_include=[BlogComment, Account]
+
+)
+print("BlogComment111")
+
+crud_route_child2 = generic_sql_crud_router_builder(
+    db_model=BlogComment,
+    prefix="/blog_comment",
+    tags=["blog_comment"],
+    foreign_include=[BlogPost]
+
+)
+
+app = FastAPI()
+[app.include_router(i) for i in [crud_route_parent, crud_route_child1, crud_route_child2]]
+
+
+@app.get("/", tags=["child"])
+async def root():
+    return {"message": "Hello World"}
+
+
+import uvicorn
+
+uvicorn.run(app, host="0.0.0.0", port=8002, debug=False)

src/fastapi_quickcrud/test.py

@@ -3,54 +3,43 @@
 from sqlalchemy import *
 from sqlalchemy.orm import *
 from fastapi_quickcrud.crud_router import generic_sql_crud_router_builder
+from fastapi_quickcrud.misc.covert_model import convert_table_to_model
 
 Base = declarative_base()
 
-class Account(Base):
-    __tablename__ = "account"
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    blog_post = relationship("BlogPost", back_populates="account")
-class BlogPost(Base):
-    __tablename__ = "blog_post"
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    account_id = Column(Integer, ForeignKey("account.id"), nullable=False)
-    account = relationship("Account", back_populates="blog_post")
-    blog_comment = relationship("BlogComment", back_populates="blog_post")
-
-class BlogComment(Base):
-    __tablename__ = "blog_comment"
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    blog_id = Column(Integer, ForeignKey("blog_post.id"), nullable=False)
-    blog_post = relationship("BlogPost", back_populates="blog_comment")
-
-
-crud_route_parent = crud_router_builder(
-    db_model=Account,
-    prefix="/account",
-    tags=["account"],
-    foreign_include=[BlogComment,BlogPost]
-
-)
-print("BlogPost222")
-
-crud_route_child1 = generic_sql_crud_router_builder(
-    db_model=BlogPost,
-    prefix="/blog_post",
-    tags=["blog_post"],
-    foreign_include = [BlogComment]
-
-)
-print("BlogComment111")
-
-crud_route_child2 = generic_sql_crud_router_builder(
-    db_model=BlogComment,
-    prefix="/blog_comment",
-    tags=["blog_comment"]
-
+association_table = Table('association', Base.metadata,
+    Column('left_id', ForeignKey('left.id'), primary_key=True),
+    Column('right_id', ForeignKey('right.id'), primary_key=True)
 )
 
+class Parent(Base):
+    __tablename__ = 'left'
+    id = Column(Integer, primary_key=True)
+    children = relationship(
+        "Child",
+        secondary=association_table,
+        back_populates="parents")
+
+class Child(Base):
+    __tablename__ = 'right'
+    id = Column(Integer, primary_key=True)
+    parents = relationship(
+        "Parent",
+        secondary=association_table,
+        back_populates="children")
+
+crud_route_1 = crud_router_builder(db_model=Parent,
+                                   prefix="/parent",
+                                   tags=["Parent"],
+                                   foreign_include=[Child]
+                                   )
+crud_route_2 = crud_router_builder(db_model=Child,
+                                   prefix="/child",
+                                   tags=["Child"],
+                                   foreign_include=[Parent]
+                                   )
 app = FastAPI()
-[app.include_router(i) for i in [crud_route_parent, crud_route_child1, crud_route_child2]]
+[app.include_router(i) for i in [crud_route_1, crud_route_2]]
 
 @app.get("/", tags=["child"])
 async def root():

tests/test_implementations/test_memory_sqlalchemy/api_test/foreign_tree/__init__.py

@@ -0,0 +1,112 @@
+import json
+
+from fastapi import FastAPI
+from sqlalchemy import Column, Integer, \
+    ForeignKey, Table, String
+from sqlalchemy.orm import declarative_base, sessionmaker, relationship
+
+from src.fastapi_quickcrud.misc.type import SqlType
+from src.fastapi_quickcrud.crud_router import crud_router_builder
+
+app = FastAPI()
+
+Base = declarative_base()
+metadata = Base.metadata
+
+from sqlalchemy import create_engine
+
+from sqlalchemy.pool import StaticPool
+engine = create_engine('sqlite://', echo=True,
+                       connect_args={"check_same_thread": False}, pool_recycle=7200, poolclass=StaticPool)
+session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+def get_transaction_session():
+    try:
+        db = session()
+        yield db
+    finally:
+        db.close()
+
+
+class User(Base):
+    __tablename__ = 'test_users'
+    id = Column(Integer, primary_key=True, autoincrement=True, unique=True)
+    name = Column(String, nullable=False)
+    email = Column(String, nullable=False)
+
+
+friend = Table(
+    'test_friend', Base.metadata,
+    Column('id', ForeignKey('test_users.id', ondelete='CASCADE', onupdate='CASCADE'), nullable=False),
+    Column('friend_name', String, nullable=False)
+)
+
+crud_route_1 = crud_router_builder(db_session=get_transaction_session,
+                                       db_model=User,
+                                   prefix="/user",
+                                   tags=["User"],
+                                   foreign_include=[friend]
+                                   )
+crud_route_2 = crud_router_builder(db_session=get_transaction_session,
+                                       db_model=friend,
+                                   prefix="/friend",
+                                   tags=["friend"],
+                                   foreign_include=[User]
+                                   )
+
+from starlette.testclient import TestClient
+
+[app.include_router(i) for i in
+ [crud_route_1,crud_route_2]]
+
+client = TestClient(app)
+
+
+def test_():
+    headers = {
+        'accept': '*/*',
+        'Content-Type': 'application/json',
+    }
+    data = '[{"id": 1,"name": "string","email": "string"}]'
+    response = client.post('/user', headers=headers, data=data)
+    assert response.status_code == 201
+    assert response.json() == [{"id": 1,"name": "string","email": "string"}]
+
+
+    data  =' [{"id": 1,"friend_name": "string"}]'
+    response = client.post('/friend', headers=headers, data = data)
+    assert response.status_code == 201
+    assert response.json() == [
+  {
+    "id": 1,
+    "friend_name": "string"
+  }
+]
+
+    response = client.get('/friend?join_foreign_table=test_users', headers=headers)
+    assert response.status_code == 200
+    assert response.json() == [
+  {
+    "test_users_foreign": [
+      {
+        "id": 1,
+        "name": "string",
+        "email": "string"
+      }
+    ],
+    "id": 1,
+    "friend_name": "string"
+  }
+]
+
+
+
+def setup_module(module):
+    User.__table__.create(module.engine, checkfirst=True)
+    friend.create(module.engine, checkfirst=True)
+
+
+def teardown_module(module):
+    friend.drop(engine, checkfirst=True)
+    User.__table__.drop(engine, checkfirst=True)