Added sql.upload_pm_patch_catalog, added prints to get_patch_catalog,…

… fixed bug in sql.db_connect not recognizing db_type=sqlite3

docs/sql.md

@@ -139,6 +139,7 @@ The final optional parameter is ```table_name```. If you want to specify a custo
 | ```upload_pm_patches``` | Patch Management | ```pm.get_patches()``` | ```pm_patches``` |
 | ```upload_pm_assets``` | Patch Management | ```pm.get_assets()``` | ```pm_assets``` |
 | ```upload_pm_assetids_to_uuids``` | Patch Management | ```pm.lookup_host_uuids()``` | ```pm_assetids_to_uuids``` |
+| ```upload_pm_patch_catalog``` | Patch Management | ```pm.get_patch_catalog()``` | ```pm_patch_catalog``` |
 | ```upload_cert_certs``` | Certificate View | ```cert.list_certs()``` | ```cert_certs``` for certificates and ```cert_assets``` for assets (key = certs.id -> assets.certId) |
 
 

qualysdk/pm/data_classes/CatalogPatch.py

@@ -68,7 +68,7 @@ class CatalogPatch(BaseClass):
     notification: str = None
     cve: BaseList[str] = None
     architecture: BaseList[str] = None
-    packageDetails: BaseList[str] = None
+    packageDetails: BaseList[PackageDetail] = None
     patchFeedProviderId: int = None
     syncDateTime: Union[int, datetime] = None
     vendorPatchId: Union[str, int] = None

qualysdk/pm/patchcatalog.py

@@ -53,6 +53,7 @@ def get_patch_catalog(auth: TokenAuth, patchId: str, platform: Literal["windows"
         params["attributes"] = kwargs["attributes"]
 
     results = BaseList()
+    pulled = 0
 
     # Set up chunking
     while True:
@@ -82,4 +83,8 @@ def get_patch_catalog(auth: TokenAuth, patchId: str, platform: Literal["windows"
         for catalog_entry in j:
             results.append(CatalogPatch(**catalog_entry))
 
+        pulled += 1
+        if pulled % 5 == 0:
+            print(f"Pulled {pulled} chunks of 1K patch catalog entries")
+
     return results

qualysdk/sql/__init__.py

@@ -73,5 +73,6 @@
     upload_pm_patches,
     upload_pm_assets,
     upload_pm_assetids_to_uuids,
+    upload_pm_patch_catalog,
 )
 from .cert import upload_cert_certs

qualysdk/sql/base.py

@@ -18,7 +18,7 @@ def db_connect(
     username: str = None,
     password: str = None,
     trusted_cnxn: bool = False,
-    db_type: Literal["mssql", "mysql", "postgresql", "sqlite"] = "mssql",
+    db_type: Literal["mssql", "mysql", "postgresql", "sqlite", "sqlite3"] = "mssql",
     port: int = 1433,
 ) -> Connection:
     """
@@ -31,15 +31,15 @@ def db_connect(
         username (str): The username to use to connect to the database.
         password (str): The password to use to connect to the database.
         trusted_cnxn (bool): If True, use trusted connection on MSSQL. If False, use username and password. Defaults to False.
-        db_type (str): The type of database to connect to. Defaults to 'mssql'. Options are 'mssql', 'mysql', 'postgresql', 'sqlite'.
+        db_type (str): The type of database to connect to. Defaults to 'mssql'. Options are 'mssql', 'mysql', 'postgresql', 'sqlite', 'sqlite3'.
         port (int): The port to connect to the database on. Defaults to 1433.
 
     Returns:
         Connection: The Connection object to the SQL database.
     """
 
     # Check if user AND password are provided, or trusted connection is used:
-    if not (username and password) and not trusted_cnxn and db_type != "sqlite":
+    if not (username and password) and not trusted_cnxn and db_type not in ["sqlite", "sqlite3"]:
         raise ValueError(
             "You must provide a username and password, or use trusted connection."
         )
@@ -57,9 +57,7 @@ def db_connect(
             conn_str = f"mysql+pymysql://{username}:{password}@{host}:{port}/{db}"
         case "postgresql":
             conn_str = f"postgresql+psycopg2://{username}:{password}@{host}:{port}/{db}"
-        case "sqlite":
-            conn_str = f"sqlite:///{db}"
-        case "sqlite3":
+        case "sqlite"|"sqlite3":
             conn_str = f"sqlite:///{db}"
         case _:
             raise ValueError("Database type not supported.")

qualysdk/sql/pm.py

@@ -575,3 +575,71 @@ def upload_pm_assetids_to_uuids(
 
     # Upload the data:
     return upload_data(df, table_name, cnxn, COLS, override_import_dt)
+
+def upload_pm_patch_catalog(
+    patches: BaseList,
+    cnxn: Connection,
+    table_name: str = "pm_patch_catalog",
+    override_import_dt: datetime = None,
+) -> int:
+    """
+    Upload results from ```pm.get_patch_catalog```
+    to a SQL database.
+
+    Args:
+        patches (BaseList): A BaseList of CatalogPatch objects.
+        cnxn (Connection): The Connection object to the SQL database.
+        table_name (str): The name of the table to upload to. Defaults to "pm_patch_catalog".
+        override_import_dt (datetime): If provided, will override the import_datetime column with this value.
+
+    Returns:
+        int: The number of rows uploaded.
+    """
+
+    COLS = {
+        "patchId": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "id": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "title": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "type": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "appFamily": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "vendor": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "product": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "platform": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "kb": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "isSuperseded": types.Boolean(),
+        "isSecurity": types.Boolean(),
+        "isRollback": types.Boolean(),
+        "servicePack": types.Boolean(),
+        "advisory": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "vendorlink": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "osIdentifier": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "advisoryLink": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "deleted": types.Boolean(),
+        "rebootRequired": types.Boolean(),
+        "vendorSeverity": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "description": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "qid": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "enabled": types.Boolean(),
+        "downloadMethod": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "supportedOs": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "supersedes": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "notification": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "cve": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "architecture": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "packageDetails": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "patchFeedProviderId": types.Integer(),
+        "syncDateTime": types.DateTime(),
+        "vendorPatchId": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "modifiedDate": types.DateTime(),
+        "publishedDate": types.DateTime(),
+        "category": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "isEsuPatch": types.Boolean(),
+        "supersededBy": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "bulletin": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+    }
+
+    # Prepare the dataclass for insertion:
+    df = DataFrame([prepare_dataclass(patch) for patch in patches])
+
+    # Upload the data:
+    return upload_data(df, table_name, cnxn, COLS, override_import_dt)