Create Org

d2x/auth/sf/auth_url.py

@@ -12,6 +12,7 @@
 from rich.panel import Panel
 from rich.progress import Progress, SpinnerColumn, TextColumn
 from rich.table import Table
+from simple_salesforce import Salesforce
 
 # Local imports
 from d2x.models.sf.auth import (
@@ -23,7 +24,7 @@
     SfdxAuthUrlModel,
 )
 from d2x.ux.gh.actions import summary as gha_summary, output as gha_output
-from d2x.models.sf.org import SalesforceOrgInfo
+from d2x.models.sf.org import SalesforceOrgInfo, ScratchOrg
 from d2x.base.types import CLIOptions
 from d2x.api.gh import (
     set_environment_variable,
@@ -62,7 +63,7 @@ def exchange_token(org_info: SalesforceOrgInfo, cli_options: CLIOptions):
 
             # Create debug info
             debug_info = TokenExchangeDebug(
-                url=f"https://{org_info.full_domain}{token_url_path}",
+                url=f"https://{org_info.org.full_domain}{token_url_path}",
                 method="POST",
                 headers=headers,
                 request=token_request,
@@ -71,8 +72,8 @@ def exchange_token(org_info: SalesforceOrgInfo, cli_options: CLIOptions):
             console.print(debug_info.to_table())
 
             # Make request
-            progress.add_task(f"Connecting to {org_info.full_domain}...", total=None)
-            conn = http.client.HTTPSConnection(org_info.full_domain)
+            progress.add_task(f"Connecting to {org_info.org.full_domain}...", total=None)
+            conn = http.client.HTTPSConnection(org_info.org.full_domain)
 
             task = progress.add_task("Exchanging tokens...", total=None)
             conn.request("POST", token_url_path, body, headers)
@@ -116,7 +117,7 @@ def exchange_token(org_info: SalesforceOrgInfo, cli_options: CLIOptions):
 
             # Display success
             success_panel = Panel(
-                f"[green]Successfully authenticated to {org_info.full_domain}\n"
+                f"[green]Successfully authenticated to {org_info.org.full_domain}\n"
                 f"[blue]Token Details:[/]\n"
                 f"  Issued At: {token_response.issued_at.strftime('%Y-%m-%d %H:%M:%S')}\n"
                 f"  Expires At: {token_response.expires_at.strftime('%Y-%m-%d %H:%M:%S')}\n"
@@ -148,9 +149,56 @@ def exchange_token(org_info: SalesforceOrgInfo, cli_options: CLIOptions):
             raise
 
 
+def create_scratch_org(org_info: ScratchOrg, cli_options: CLIOptions):
+    """Create a scratch org using simple-salesforce"""
+    console = cli_options.console
+    with Progress(
+        SpinnerColumn(),
+        TextColumn("[progress.description]{task.description}"),
+        console=console,
+        transient=True,
+    ) as progress:
+        try:
+            progress.add_task("Creating scratch org...", total=None)
+
+            # Authenticate to Salesforce using simple-salesforce
+            sf = Salesforce(
+                username=org_info.auth_info.client_id,
+                password=org_info.auth_info.client_secret.get_secret_value(),
+                security_token=org_info.auth_info.refresh_token,
+                domain=org_info.full_domain,
+            )
+
+            # Create scratch org
+            result = sf.restful("sobjects/ScratchOrg", method="POST", json=org_info.dict())
+
+            # Display success
+            success_panel = Panel(
+                f"[green]Successfully created scratch org\n"
+                f"[blue]Org Details:[/]\n"
+                f"  Org ID: {result['id']}\n"
+                f"  Status: {result['status']}\n"
+                f"  Expiration Date: {result['expirationDate']}",
+                title="[green]Scratch Org Creation Success",
+                border_style="green",
+            )
+            console.print(success_panel)
+
+            return result
+
+        except Exception as e:
+            error_panel = Panel(
+                f"[red]Error: {str(e)}",
+                title="[red]Scratch Org Creation Failed",
+                border_style="red",
+            )
+            console.print(error_panel)
+            raise
+
+
 def get_full_domain(org_info: SalesforceOrgInfo) -> str:
     """Construct the full domain from SalesforceOrgInfo."""
-    return org_info.full_domain.rstrip("/")
+    return org_info.org.full_domain.rstrip("/")
 
 
 def main(cli_options: CLIOptions):

d2x/cli/main.py

@@ -7,6 +7,8 @@
 from typing import Optional
 from importlib.metadata import version, PackageNotFoundError
 from d2x.env.gh import set_environment_variable, get_environment_variable, set_environment_secret, get_environment_secret
+from d2x.auth.sf.auth_url import create_scratch_org, exchange_token
+from d2x.models.sf.org import SalesforceOrgInfo, ScratchOrg
 
 # Disable rich_click's syntax highlighting
 click.SHOW_ARGUMENTS = False
@@ -80,6 +82,68 @@ def url(output_format: OutputFormatType, debug: bool):
             raise
 
 
+@sf.group()
+def org():
+    """Salesforce org commands"""
+    pass
+
+
+@org.command()
+@common_options
+def create(output_format: OutputFormatType, debug: bool):
+    """Create a Salesforce scratch org"""
+    cli_options = CLIOptions(output_format=output_format, debug=debug)
+    try:
+        # Assuming org_info is obtained from somewhere, e.g., environment variable or input
+        org_info = ScratchOrg(
+            org_type="scratch",
+            domain_type="my",
+            full_domain="example.my.salesforce.com",
+            auth_info=AuthInfo(
+                client_id="your_client_id",
+                client_secret="your_client_secret",
+                refresh_token="your_refresh_token",
+                instance_url="https://example.my.salesforce.com",
+            ),
+        )
+        create_scratch_org(org_info, cli_options)
+    except:
+        if debug:
+            type, value, tb = sys.exc_info()
+            pdb.post_mortem(tb)
+        else:
+            raise
+
+
+@org.command()
+@common_options
+def exchange(output_format: OutputFormatType, debug: bool):
+    """Exchange auth code for an OAuth grant"""
+    cli_options = CLIOptions(output_format=output_format, debug=debug)
+    try:
+        # Assuming org_info is obtained from somewhere, e.g., environment variable or input
+        org_info = SalesforceOrgInfo(
+            auth_info=AuthInfo(
+                client_id="your_client_id",
+                client_secret="your_client_secret",
+                refresh_token="your_refresh_token",
+                instance_url="https://example.my.salesforce.com",
+            ),
+            org=ScratchOrg(
+                org_type="scratch",
+                domain_type="my",
+                full_domain="example.my.salesforce.com",
+            ),
+        )
+        exchange_token(org_info, cli_options)
+    except:
+        if debug:
+            type, value, tb = sys.exc_info()
+            pdb.post_mortem(tb)
+        else:
+            raise
+
+
 @d2x_cli.group()
 def env():
     """Environment commands"""

d2x/models/sf/org.py

@@ -1,5 +1,5 @@
-from typing import Optional, Literal
-from pydantic import Field
+from typing import Optional, Literal, Union
+from pydantic import Field, BaseModel
 from d2x.base.models import CommonBaseModel
 from d2x.models.sf.auth import AuthInfo, DomainType, OrgType
 
@@ -23,12 +23,9 @@
 PodType = Literal["cs", "db", None]
 
 
-class SalesforceOrgInfo(CommonBaseModel):
-    """Structured information about a Salesforce org."""
+class BaseSalesforceOrg(BaseModel):
+    """Base model for Salesforce orgs with a pydantic discriminator."""
 
-    auth_info: AuthInfo = Field(
-        ..., description="Authentication information for the Salesforce org."
-    )
     org_type: OrgType = Field(..., description="Type of the Salesforce org.")
     domain_type: DomainType = Field(
         ..., description="Type of domain for the Salesforce org."
@@ -44,10 +41,79 @@ class SalesforceOrgInfo(CommonBaseModel):
     pod_number: Optional[str] = Field(None, description="Pod number if applicable.")
     pod_type: Optional[PodType] = Field(None, description="Pod type if applicable.")
 
+    class Config:
+        use_enum_values = True
+        discriminator = "org_type"
+
+
+class ProductionOrg(BaseSalesforceOrg):
+    """Model for Production Salesforce orgs."""
+
+    org_type: Literal[OrgType.PRODUCTION] = Field(
+        default=OrgType.PRODUCTION, description="Type of the Salesforce org."
+    )
+    instance_url: str = Field(..., description="Instance URL of the production org.")
+    created_date: str = Field(..., description="Creation date of the production org.")
+    last_modified_date: str = Field(..., description="Last modified date of the production org.")
+    status: str = Field(..., description="Status of the production org.")
+
+
+class TrialOrg(BaseSalesforceOrg):
+    """Model for Trial Salesforce orgs."""
+
+    org_type: Literal[OrgType.DEMO] = Field(
+        default=OrgType.DEMO, description="Type of the Salesforce org."
+    )
+    expiration_date: Optional[str] = Field(
+        None, description="Expiration date of the trial org."
+    )
+    instance_url: str = Field(..., description="Instance URL of the trial org.")
+    created_date: str = Field(..., description="Creation date of the trial org.")
+    last_modified_date: str = Field(..., description="Last modified date of the trial org.")
+    status: str = Field(..., description="Status of the trial org.")
+
+
+class SandboxOrg(BaseSalesforceOrg):
+    """Model for Sandbox Salesforce orgs."""
+
+    org_type: Literal[OrgType.SANDBOX] = Field(
+        default=OrgType.SANDBOX, description="Type of the Salesforce org."
+    )
+    instance_url: str = Field(..., description="Instance URL of the sandbox org.")
+    created_date: str = Field(..., description="Creation date of the sandbox org.")
+    last_modified_date: str = Field(..., description="Last modified date of the sandbox org.")
+    status: str = Field(..., description="Status of the sandbox org.")
+
+
+class ScratchOrg(BaseSalesforceOrg):
+    """Model for Scratch Salesforce orgs."""
+
+    org_type: Literal[OrgType.SCRATCH] = Field(
+        default=OrgType.SCRATCH, description="Type of the Salesforce org."
+    )
+    expiration_date: Optional[str] = Field(
+        None, description="Expiration date of the scratch org."
+    )
+    instance_url: str = Field(..., description="Instance URL of the scratch org.")
+    created_date: str = Field(..., description="Creation date of the scratch org.")
+    last_modified_date: str = Field(..., description="Last modified date of the scratch org.")
+    status: str = Field(..., description="Status of the scratch org.")
+
+
+class SalesforceOrgInfo(CommonBaseModel):
+    """Structured information about a Salesforce org."""
+
+    auth_info: AuthInfo = Field(
+        ..., description="Authentication information for the Salesforce org."
+    )
+    org: Union[ProductionOrg, TrialOrg, SandboxOrg, ScratchOrg] = Field(
+        ..., description="Salesforce org details."
+    )
+
     @property
     def is_classic_pod(self) -> bool:
         """Determine if the pod is a classic pod."""
-        return self.pod_type in ["cs", "db"]
+        return self.org.pod_type in ["cs", "db"]
 
     @property
     def is_hyperforce(self) -> bool:
@@ -57,4 +123,4 @@ def is_hyperforce(self) -> bool:
     @property
     def is_sandbox(self) -> bool:
         """Determine if the org is a sandbox."""
-        return self.org_type == OrgType.SANDBOX
+        return self.org.org_type == OrgType.SANDBOX